From 3dba2e550a442161e4f347464de30b01e0739513 Mon Sep 17 00:00:00 2001 From: dtoro Date: Sat, 11 Jul 2026 12:35:27 +0200 Subject: [PATCH] =?UTF-8?q?feat(tasks):=20phase=202=20=E2=80=94=20entity.t?= =?UTF-8?q?ouched=20events=20+=20task=E2=86=92entity=20involves=20edges?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit As the agent runs a task, record which entities each tool call references: write an idempotent task —involves→ entity relationship and publish one entity.touched event per entity (correlation_id = session, data {slug,tool}). Extracted from tool ARGS only — never results — so a bulk fleet query can't drag every entity into the task graph; bulk/no-slug tools stay silent. Emitted from the nomos agent loop rather than the shared MCP wrapper, which has no session id. The involves edges make a task's graph neighborhood its involved-entity set (queryable via get_relations) — the substrate for the knowledge loop; the events are the live pulse the context panel consumes in phase 6. Verified end-to-end on the local stack: a chat referencing lxc:caddy/lxc:gitea produced entity.touched on the browser SSE stream with slug+tool+correlation, and exactly one involves edge per entity despite repeated touches. Co-Authored-By: Claude Opus 4.8 --- cmd/nomos/agent.go | 5 +++ cmd/nomos/store.go | 80 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+) diff --git a/cmd/nomos/agent.go b/cmd/nomos/agent.go index 8930aad..f7e9afb 100644 --- a/cmd/nomos/agent.go +++ b/cmd/nomos/agent.go @@ -392,6 +392,11 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s a.store.linkExecution(ctx, execID, sessionID) } + // Record which entities this task touched (task —involves→ entity) + // and pulse them on the live context panel. Args only — never + // results — so a bulk query doesn't drag the whole fleet in. + a.store.recordTouched(ctx, sessionID, tc.Function.Name, args) + emit(agentEvent{ Type: "tool_result", Data: map[string]any{"name": tc.Function.Name, "result": result, "id": tc.ID}, diff --git a/cmd/nomos/store.go b/cmd/nomos/store.go index 4a00f80..8b41e6c 100644 --- a/cmd/nomos/store.go +++ b/cmd/nomos/store.go @@ -5,8 +5,12 @@ import ( "encoding/json" "fmt" "log/slog" + "regexp" + "strings" "time" + "github.com/dtoro/oikos/internal/db/sqlcgen" + "github.com/dtoro/oikos/internal/observability" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgxpool" ) @@ -280,6 +284,82 @@ func (s *store) linkExecution(ctx context.Context, execID uuid.UUID, sessionID s VALUES ($1, $2) ON CONFLICT (execution_id) DO NOTHING`, execID, sessionID) } +// taskSlugRe matches an entity slug: a lowercase type prefix then colon- +// separated segments (host:hubris, lxc:caddy, check:ping:8cf). Mirrors the +// frontend SessionGraph regex so the panel and the involves-graph agree on +// what counts as an entity reference. +var taskSlugRe = regexp.MustCompile(`[a-z][a-z-]*:[a-z0-9][a-z0-9._/-]*(?::[a-z0-9._/-]+)*`) + +// touchExcludedTypes are entity types too noisy to record as task involvement: +// a health question names dozens of check:… slugs, executions/tasks are +// bookkeeping, not things the task "worked on". +var touchExcludedTypes = map[string]bool{"check": true, "execution": true, "task": true} + +// recordTouched links the task to every entity referenced in a tool call's +// args (task —involves→ entity) and publishes one entity.touched event per +// entity so the live context panel can pulse it. Best-effort: it never blocks +// or fails the tool call. Only args are inspected — what the agent chose to act +// on — never results, since a single bulk query result would otherwise pull the +// whole fleet into the task's graph. +func (s *store) recordTouched(ctx context.Context, sessionID, toolName string, args map[string]any) { + if s == nil || sessionID == "" || sessionID == "ephemeral" || len(args) == 0 { + return + } + slugs := map[string]struct{}{} + collectTaskSlugs(args, slugs) + if len(slugs) == 0 { + return + } + var taskEntityID uuid.UUID + if err := s.pool.QueryRow(ctx, + `SELECT entity_id FROM agent_sessions WHERE id = $1`, sessionID).Scan(&taskEntityID); err != nil || taskEntityID == uuid.Nil { + return // no task entity to anchor edges on + } + q := sqlcgen.New(s.pool) + for slug := range slugs { + var entID uuid.UUID + var etype string + if err := s.pool.QueryRow(ctx, + `SELECT id, type FROM entities WHERE slug = $1`, slug).Scan(&entID, &etype); err != nil { + continue // unknown slug — skip + } + if touchExcludedTypes[etype] || entID == taskEntityID { + continue + } + // Idempotent involves edge (task → entity), same guard as upsert_knowledge. + s.pool.Exec(ctx, ` + INSERT INTO relationships (source_id, target_id, type, attributes, valid_from) + SELECT $1, $2, 'involves', '{"by":"nomos"}'::jsonb, now() + WHERE NOT EXISTS ( + SELECT 1 FROM relationships + WHERE source_id = $1 AND target_id = $2 AND type = 'involves' AND valid_to IS NULL)`, + taskEntityID, entID) + // Live pulse for the panel. correlation_id = sessionID lets the frontend + // filter to the active task. + _ = observability.Event(ctx, q, "entity.touched", &entID, "info", "nomos", sessionID, + map[string]any{"slug": slug, "tool": toolName}) + } +} + +// collectTaskSlugs recursively pulls entity slugs out of tool-call args, +// mirroring the frontend's collectSlugs so both sides see the same references. +func collectTaskSlugs(v any, out map[string]struct{}) { + switch t := v.(type) { + case string: + for _, m := range taskSlugRe.FindAllString(t, -1) { + out[strings.TrimRight(m, ".,;)]")] = struct{}{} + } + case []any: + for _, e := range t { + collectTaskSlugs(e, out) + } + case map[string]any: + for _, e := range t { + collectTaskSlugs(e, out) + } + } +} + // pendingContinuation is one finished execution whose result hasn't yet been // fed back to its originating session. type pendingContinuation struct {