dtoro/oikos
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix(mcp): write targets/involves relationship edges when executions are created

Browse Source 
Executions were being created with no outgoing edges to what they acted
on or which task/session drove them, silently starving the graph of new
data going forward — found during this session's DB audit, which had to
backfill 245+25 missing targets/involves edges for existing executions.
This closes the gap at the source: every execution now gets a
target-->targets-->execution edge, and (when the caller supplies a
session/task) a task-->involves-->execution edge, both idempotent
(NOT EXISTS guards) so retries/backfills don't duplicate.

Two call sites: the deduped systemctl/apt_upgrade/pct_create fast path
and the general classifyAndGate path.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
authentik Default Admin
2026-07-13 10:45:44 +02:00
parent 335fa67d55
commit 62a8ec1d8d
1 changed files with 34 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
internal/mcp/server.go
View File

@@ -428,6 +428,23 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server {
}
pool.Exec(ctx, `INSERT INTO executions (entity_id, target_entity_id, action, risk_class, status, correlation_id, agent_id) VALUES ($1, $2, $3, 'reversible_low', 'running', $4, $5) ON CONFLICT DO NOTHING`,
id, targetID, action+":"+params, correlationID, agentID)
pool.Exec(ctx, `
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
SELECT $1, $2, 'targets', '{"by":"nomos"}'::jsonb, now()
WHERE NOT EXISTS (
SELECT 1 FROM relationships
WHERE source_id = $1 AND target_id = $2 AND type = 'targets' AND valid_to IS NULL)`,
id, targetID)
if sessionID != "" {
pool.Exec(ctx, `
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
SELECT t.id, $1, 'involves', '{"by":"nomos"}'::jsonb, now()
FROM entities t WHERE t.slug = $2
AND NOT EXISTS (
SELECT 1 FROM relationships
WHERE source_id = t.id AND target_id = $1 AND type = 'involves' AND valid_to IS NULL)`,
id, "task:"+sessionID)
}
// Execute reversible actions immediately. restart/pct_exec/systemctl
// (outside enable/disable) never reach here — they're routed through
@@ -1397,6 +1414,23 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.
}
pool.Exec(ctx, `INSERT INTO executions (entity_id, target_entity_id, action, risk_class, status, correlation_id, agent_id) VALUES ($1, $2, $3, $4, 'running', $5, $6) ON CONFLICT DO NOTHING`,
id, targetID, actionCol, riskClass, correlationID, agentID)
pool.Exec(ctx, `
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
SELECT $1, $2, 'targets', '{"by":"nomos"}'::jsonb, now()
WHERE NOT EXISTS (
SELECT 1 FROM relationships
WHERE source_id = $1 AND target_id = $2 AND type = 'targets' AND valid_to IS NULL)`,
id, targetID)
if sessionID != "" {
pool.Exec(ctx, `
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
SELECT t.id, $1, 'involves', '{"by":"nomos"}'::jsonb, now()
FROM entities t WHERE t.slug = $2
AND NOT EXISTS (
SELECT 1 FROM relationships
WHERE source_id = t.id AND target_id = $1 AND type = 'involves' AND valid_to IS NULL)`,
id, "task:"+sessionID)
}
if riskClass == policy.RiskReadOnly {
host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug)