From 62a8ec1d8de44394e08691c0ecef2feb3b66567a Mon Sep 17 00:00:00 2001 From: dtoro Date: Mon, 13 Jul 2026 10:45:44 +0200 Subject: [PATCH] fix(mcp): write targets/involves relationship edges when executions are created MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- internal/mcp/server.go | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 0ad6b6d..4f50439 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -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)