From b98d7c24bf6fe7f9f2c44ea1e449cc1b1da880d1 Mon Sep 17 00:00:00 2001 From: dtoro Date: Sun, 16 Aug 2026 12:29:59 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Phase=209=20gaps=20closed=20=E2=80=94?= =?UTF-8?q?=20ApprovalService.Decide=20convergence,=20execlog=20fold,=20ex?= =?UTF-8?q?ecworker=20poller?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ApprovalService (core/app/approval.go) + ApprovalRepo (postgres adapter) with full decide transaction: HMAC token verify, approval flip, execution un-gate, session-scoped window keys (+session suffix matching GovernanceStore gate), nomos session flip, audit+event on failure abort. httpapi DecideApproval now a thin presenter delegating to the service. ListPending payload format fixed (json.Unmarshal not raw-wrap). - execlog folded into postgres adapter: internal/execlog deleted, NewExecutionLog / ReadExecutionLog live in the db package, callers updated (mcp, httpapi). - execworker poller over ExecutionService.DispatchQueued: advisory lock leak fixed (defer/recover per execution), correlation_id preserved via Finalize event emission (ExecRunRepo.Finalize now emits execution.{status} with correlation_id from the row). - Phase 8 session export-rename completed: Store, New, and all 53 methods exported; cmd/nomos/ agent.go fixed to use session.PendingContinuation etc. - Coverage gates: ExecutionService.Submit 93.1%, PolicyService.Decide 100%. - Plans index updated, VERSION bumped to 0.36.0. --- VERSION | 2 +- cmd/nomos/agent.go | 67 +++- cmd/nomos/continue_test.go | 16 +- cmd/nomos/server.go | 5 +- cmd/nomos/tasks.go | 8 +- cmd/oikos/main.go | 31 +- internal/adapters/postgres/approvals.go | 306 ++++++++++++++++++ .../{execlog => adapters/postgres}/execlog.go | 51 ++- internal/adapters/postgres/governance.go | 23 +- internal/core/app/approval.go | 65 ++++ internal/core/app/approval_test.go | 114 +++++++ internal/core/app/execution.go | 34 +- internal/core/app/execution_test.go | 132 ++++++++ internal/core/ports/governance.go | 19 +- internal/core/ports/portstest/fakes.go | 63 +++- internal/execworker/worker.go | 157 +++------ internal/httpapi/actuator.go | 5 +- internal/httpapi/api_test.go | 5 +- internal/httpapi/approvals.go | 215 +++--------- internal/httpapi/execution_logs.go | 4 +- internal/httpapi/server.go | 21 +- internal/mcp/discover.go | 21 +- internal/mcp/server.go | 9 +- internal/nomos/assent/assent_test.go | 32 +- internal/nomos/session/store.go | 222 +++++++------ internal/nomos/session/store_test.go | 130 ++++---- plans/index.md | 4 +- 27 files changed, 1161 insertions(+), 600 deletions(-) create mode 100644 internal/adapters/postgres/approvals.go rename internal/{execlog => adapters/postgres}/execlog.go (69%) create mode 100644 internal/core/app/approval.go create mode 100644 internal/core/app/approval_test.go diff --git a/VERSION b/VERSION index 731b95d7..93d4c1ef 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.35.1 +0.36.0 diff --git a/cmd/nomos/agent.go b/cmd/nomos/agent.go index 0bde6b55..b7968816 100644 --- a/cmd/nomos/agent.go +++ b/cmd/nomos/agent.go @@ -1,6 +1,7 @@ package main import ( + "bytes" "context" "encoding/json" "fmt" @@ -153,6 +154,40 @@ gated mutations through run. Be concise. Prefer tools over guessing.` // explicit typed confirmation regardless of the window. const assentWindowDuration = 30 * time.Minute +// approveExecution calls the oikos API to approve a pending execution. +// Returns true + new status on success, false on any error. +func (a *agent) approveExecution(ctx context.Context, execID string) (bool, string, error) { + if a.apiBase == "" || a.apiToken == "" { + return false, "", fmt.Errorf("nomos: apiBase or apiToken not configured") + } + body, _ := json.Marshal(map[string]string{"status": "approved"}) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + a.apiBase+"/api/v1/approvals/"+execID+"/decision", bytes.NewReader(body)) + if err != nil { + return false, "", fmt.Errorf("nomos: create approval request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+a.apiToken) + req.Header.Set("Content-Type", "application/json") + resp, err := a.httpClient.Do(req) + if err != nil { + return false, "", fmt.Errorf("nomos: approval request failed: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusNotFound { + return false, "", fmt.Errorf("nomos: approval %s not found", execID) + } + if resp.StatusCode != http.StatusOK { + return false, "", fmt.Errorf("nomos: approval %s returned %d", execID, resp.StatusCode) + } + var result struct { + Status string `json:"status"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return false, "", fmt.Errorf("nomos: decode approval response: %w", err) + } + return true, result.Status, nil +} + // openAssentWindow records an active assent window in autonomy_settings so // the MCP run tool (separate process) can check it before requiring approval // for config_mutation commands. Key is scoped to this agent's UUID AND this @@ -295,12 +330,12 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s // isTypedConfirmation ("I confirm ...", per SOUL.md's guidance for what // to ask the operator to type). pending := assent.ExtractPendingApprovals(func() []string { - texts := make([]string, len(lastAssistantCalls)) - for i, c := range lastAssistantCalls { - texts[i] = c.resultText() - } - return texts - }()) + texts := make([]string, len(lastAssistantCalls)) + for i, c := range lastAssistantCalls { + texts[i] = c.resultText() + } + return texts + }()) operatorAssented := assent.IsAssent(message) typedConfirm := assent.IsTypedConfirmation(message) if len(pending) > 0 && (operatorAssented || typedConfirm) { @@ -394,11 +429,11 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s var acc openai.ChatCompletionAccumulator // Capture token usage from this LLM response for activity logging. - // Previously always NULL — every agent_activity row had no token - // count. Now each tool call in this iteration gets the same total. - totalTokens := 0 + // Previously always NULL — every agent_activity row had no token + // count. Now each tool call in this iteration gets the same total. + totalTokens := 0 - for attempt := 0; attempt <= maxLLMRetries; attempt++ { + for attempt := 0; attempt <= maxLLMRetries; attempt++ { acc = openai.ChatCompletionAccumulator{} stream := a.provider.Chat.Completions.NewStreaming(ctx, params, a.reqOpts...) for stream.Next() { @@ -437,12 +472,12 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s if len(msg.ToolCalls) == 0 { if isRefusalOrEmpty(msg.Content) { - if attempt < maxLLMRetries { - slog.Warn("nomos: empty or refusal response, retrying", - "session", sessionID, "iter", i+1, "attempt", attempt+1, - "content_len", len(msg.Content), "finish_reason", finishReason) - continue - } + if attempt < maxLLMRetries { + slog.Warn("nomos: empty or refusal response, retrying", + "session", sessionID, "iter", i+1, "attempt", attempt+1, + "content_len", len(msg.Content), "finish_reason", finishReason) + continue + } // B.4: surface the real error context (finish_reason + // refusal text) instead of a generic "empty response" — // the operator can tell "content_filter — rephrase" from diff --git a/cmd/nomos/continue_test.go b/cmd/nomos/continue_test.go index 1ec06bc1..79778350 100644 --- a/cmd/nomos/continue_test.go +++ b/cmd/nomos/continue_test.go @@ -4,15 +4,17 @@ import ( "context" "testing" + "github.com/dtoro/oikos/internal/nomos/session" + "github.com/dtoro/oikos/internal/nomos/turngate" "github.com/google/uuid" ) func TestExtractExecutionIDs(t *testing.T) { // Real tool-result phrasings that should yield an execution id. pos := map[string]string{ - `"pct_create on host:strong auto-approved via assent window — execution 019f4b19-eafd-74ed-baa6-d24a27b3f52c running."`: "019f4b19-eafd-74ed-baa6-d24a27b3f52c", - `"run on lxc:caddy requires approval (risk: config_mutation) — execution 019f4af7-7eff-7723-b38c-b540b267f407 queued."`: "019f4af7-7eff-7723-b38c-b540b267f407", - `"apt_upgrade on host:hubris auto-approved via assent window — execution 019f4b58-c88c-7767-87dd-044608ced913 running."`: "019f4b58-c88c-7767-87dd-044608ced913", + `"pct_create on host:strong auto-approved via assent window — execution 019f4b19-eafd-74ed-baa6-d24a27b3f52c running."`: "019f4b19-eafd-74ed-baa6-d24a27b3f52c", + `"run on lxc:caddy requires approval (risk: config_mutation) — execution 019f4af7-7eff-7723-b38c-b540b267f407 queued."`: "019f4af7-7eff-7723-b38c-b540b267f407", + `"apt_upgrade on host:hubris auto-approved via assent window — execution 019f4b58-c88c-7767-87dd-044608ced913 running."`: "019f4b58-c88c-7767-87dd-044608ced913", } for in, want := range pos { ids := extractExecutionIDs(in) @@ -48,14 +50,14 @@ func TestExtractExecutionIDs(t *testing.T) { // return false, body never executed — when a turn is already active for the // session. continueSession relies on this so it only marks a continuation // "continued" after a turn really ran (otherwise the result is lost: marked -// continued, never re-queued by pendingContinuations). +// continued, never re-queued by PendingContinuations). // // A minimal agent with only a gate is enough: if the body ever ran, chatWith // would dereference the nil provider and panic. Returning false cleanly proves // the body was skipped. func TestResumeSession_SkipsWhenBusy(t *testing.T) { a := &agent{gate: turngate.New()} - if !a.gate.acquire("sess", 0) { + if !a.gate.Acquire("sess", 0) { t.Fatal("precondition: initial acquire should succeed on a free session") } ran := a.resumeSession(context.Background(), "sess", "note") @@ -70,9 +72,9 @@ func TestResumeSession_SkipsWhenBusy(t *testing.T) { // without reaching resumeSession's body (nil provider → panic) or markContinued. func TestContinueSession_DefersWhenBusy(t *testing.T) { a := &agent{gate: turngate.New()} - if !a.gate.acquire("sess", 0) { + if !a.gate.Acquire("sess", 0) { t.Fatal("precondition: initial acquire should succeed on a free session") } - p := pendingContinuation{ExecID: uuid.New(), SessionID: "sess", Status: "completed"} + p := session.PendingContinuation{ExecID: uuid.New(), SessionID: "sess", Status: "completed"} a.continueSession(context.Background(), p) // must not panic; must not run/mark } diff --git a/cmd/nomos/server.go b/cmd/nomos/server.go index 2565fcf8..01ac7b1b 100644 --- a/cmd/nomos/server.go +++ b/cmd/nomos/server.go @@ -17,6 +17,7 @@ import ( "github.com/dtoro/oikos/internal/nomos/session" "github.com/dtoro/oikos/internal/safego" "github.com/dtoro/oikos/internal/secrets" + "github.com/jackc/pgx/v5" ) func main() { @@ -425,7 +426,7 @@ func handleSessionsList(w http.ResponseWriter, r *http.Request, st *session.Stor limit = n } } - sessions, err := st.ListSessionsFiltered(r.Context(), ListFilter{ + sessions, err := st.ListSessionsFiltered(r.Context(), session.ListFilter{ Outcome: q.Get("outcome"), Status: q.Get("status"), EntityID: q.Get("entity_id"), @@ -692,4 +693,4 @@ func truncate(s string, n int) string { return s } return s[:n] + "..." -} \ No newline at end of file +} diff --git a/cmd/nomos/tasks.go b/cmd/nomos/tasks.go index f33bccec..eab09247 100644 --- a/cmd/nomos/tasks.go +++ b/cmd/nomos/tasks.go @@ -8,6 +8,8 @@ import ( "regexp" "strings" "time" + + "github.com/dtoro/oikos/internal/nomos/session" ) // Task tools are nomos-LOCAL, not MCP tools. They are session-scoped, and the @@ -222,7 +224,7 @@ func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args case "propose_plan": raw, _ := args["steps"].([]any) - var steps []PlanStepInput + var steps []session.PlanStepInput for _, r := range raw { m, ok := r.(map[string]any) if !ok { @@ -234,7 +236,7 @@ func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args } detail, _ := m["detail"].(string) target, _ := m["target_slug"].(string) - steps = append(steps, PlanStepInput{Title: title, Detail: detail, TargetSlug: target}) + steps = append(steps, session.PlanStepInput{Title: title, Detail: detail, TargetSlug: target}) } if len(steps) == 0 { return "error: propose_plan needs at least one step with a title", true @@ -263,7 +265,7 @@ func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args } appendedNote := "" if !hasWritebackStep { - steps = append(steps, PlanStepInput{ + steps = append(steps, session.PlanStepInput{ Title: "Write back: update_entity_attributes + create_relationship + upsert_knowledge", Detail: "Call update_entity_attributes for every entity you ran against (versions, states, counts, timestamps). Call create_relationship for any edge you discovered. Then upsert_knowledge about the affected entities (pass `about` as an array).", }) diff --git a/cmd/oikos/main.go b/cmd/oikos/main.go index a3d1b3e7..cca8f03e 100644 --- a/cmd/oikos/main.go +++ b/cmd/oikos/main.go @@ -1,7 +1,6 @@ package main import ( - "time" "context" "fmt" "log/slog" @@ -10,19 +9,20 @@ import ( "os/signal" "strings" "syscall" + "time" "github.com/google/uuid" + "github.com/dtoro/oikos/internal/adapters/postgres" + "github.com/dtoro/oikos/internal/adapters/remote" + "github.com/dtoro/oikos/internal/adapters/ssh" "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/core/app" "github.com/dtoro/oikos/internal/core/ports" - "github.com/dtoro/oikos/internal/adapters/postgres" - "github.com/dtoro/oikos/internal/adapters/remote" - intremote "github.com/dtoro/oikos/internal/remote" - "github.com/dtoro/oikos/internal/adapters/ssh" "github.com/dtoro/oikos/internal/execworker" "github.com/dtoro/oikos/internal/httpapi" "github.com/dtoro/oikos/internal/observability" + intremote "github.com/dtoro/oikos/internal/remote" "github.com/dtoro/oikos/internal/scheduler" "github.com/dtoro/oikos/internal/secrets" ) @@ -59,9 +59,9 @@ func main() { ) if sec != nil { overlays := secrets.ConfigOverlays(map[string]func(string){ - "mcp_bearer-token": func(v string) { cfg.MCPBearerToken = v }, - "api_token": func(v string) { cfg.APIToken = v }, - "oidc_client-secret": func(v string) { cfg.OIDCClientSecret = v }, + "mcp_bearer-token": func(v string) { cfg.MCPBearerToken = v }, + "api_token": func(v string) { cfg.APIToken = v }, + "oidc_client-secret": func(v string) { cfg.OIDCClientSecret = v }, }) n := secrets.OverlayConfig(ctx, sec, overlays) slog.Info("secrets resolved from Infisical", "count", n) @@ -117,7 +117,7 @@ func main() { slog.Info("all: starting api with scheduler + execution-worker in background") svc := buildAPIServices(pool) - if err := httpapi.ListenAndServe(ctx, pool, cfg, svc.entities, svc.entityRepo, svc.readModels, svc.relService, svc.provisioning, svc.seeds, svc.execSvc); err != nil { + if err := httpapi.ListenAndServe(ctx, pool, cfg, svc.entities, svc.entityRepo, svc.readModels, svc.relService, svc.provisioning, svc.seeds, svc.execSvc, svc.approvalSvc); err != nil { slog.Error("api failed", "error", err) os.Exit(1) } @@ -210,7 +210,7 @@ func runAPI(ctx context.Context, cfg config.Config) error { } svc := buildAPIServices(pool) - err = httpapi.ListenAndServe(ctx, pool, cfg, svc.entities, svc.entityRepo, svc.readModels, svc.relService, svc.provisioning, svc.seeds, svc.execSvc) + err = httpapi.ListenAndServe(ctx, pool, cfg, svc.entities, svc.entityRepo, svc.readModels, svc.relService, svc.provisioning, svc.seeds, svc.execSvc, svc.approvalSvc) if err == http.ErrServerClosed { return nil } @@ -228,6 +228,7 @@ type apiServices struct { provisioning *app.ProvisioningService seeds *app.SeedService execSvc *app.ExecutionService + approvalSvc *app.ApprovalService } func buildAPIServices(pool *db.Pool) apiServices { @@ -256,6 +257,7 @@ func buildAPIServices(pool *db.Pool) apiServices { return intremote.ResolveProxmoxHostSlug(ctx, pool, id, "") } execSvc := app.NewExecutionService(policySvc, executor, resolver, db.NewExecRunRepo(pool)) + approvalSvc := app.NewApprovalService(db.NewApprovalRepo(pool)) return apiServices{ entities: entities, @@ -265,6 +267,7 @@ func buildAPIServices(pool *db.Pool) apiServices { provisioning: provisioning, seeds: seeds, execSvc: execSvc, + approvalSvc: approvalSvc, } } @@ -394,9 +397,9 @@ func runSecretAudit(ctx context.Context, cfg config.Config) { backend := newInfisicalBackendOrFail(cfg) envValues := map[string]string{ - "mcp_bearer-token": cfg.MCPBearerToken, - "api_token": cfg.APIToken, - "oidc_client-secret": cfg.OIDCClientSecret, + "mcp_bearer-token": cfg.MCPBearerToken, + "api_token": cfg.APIToken, + "oidc_client-secret": cfg.OIDCClientSecret, } fmt.Println("key infisical env status") @@ -525,4 +528,4 @@ func runExport(ctx context.Context, cfg config.Config) error { slog.Info("exported", "file", path, "bytes", len(content)) } return nil -} \ No newline at end of file +} diff --git a/internal/adapters/postgres/approvals.go b/internal/adapters/postgres/approvals.go new file mode 100644 index 00000000..6bb7af0e --- /dev/null +++ b/internal/adapters/postgres/approvals.go @@ -0,0 +1,306 @@ +package db + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "log/slog" + "time" + + "github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen" + "github.com/dtoro/oikos/internal/core/domain" + "github.com/dtoro/oikos/internal/core/ports" + "github.com/dtoro/oikos/internal/observability" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +const ( + assentWindowDuration = 30 * time.Minute + destructiveWindowDuration = 15 * time.Minute +) + +// ApprovalRepo implements ports.ApprovalRepository over the postgres pool. +// Decide verifies the HMAC token (check-then-act), flips the approval, +// un-gates the linked execution, opens assent/destructive windows, flips +// the nomos session out of awaiting_input, and appends audit + event — one +// transaction (ADR 0016 §3.6). +type ApprovalRepo struct { + pool *Pool +} + +var _ ports.ApprovalRepository = (*ApprovalRepo)(nil) + +// NewApprovalRepo builds the approval repository. +func NewApprovalRepo(pool *Pool) *ApprovalRepo { return &ApprovalRepo{pool: pool} } + +// ListPending returns pending approvals for the given entity (or all pending +// when entityID is empty), newest first. +func (r *ApprovalRepo) ListPending(ctx context.Context, entityID domain.UUID, limit int) ([]domain.Approval, error) { + if limit <= 0 { + limit = 50 + } + if limit > 200 { + limit = 200 + } + rows, err := r.pool.Query(ctx, ` + SELECT a.entity_id, COALESCE(a.subject_entity_id::text, ''), a.action, a.risk_class, a.kind, + a.payload, a.status, COALESCE(a.token_hash, ''), a.expires_at, a.decided_at, + COALESCE(a.decided_by::text, ''), a.created_at + FROM approvals a + WHERE ($1::uuid IS NULL OR a.entity_id = $1) + AND a.status = 'pending' + ORDER BY a.created_at DESC + LIMIT $2`, uuidOrNil(entityID), limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []domain.Approval + for rows.Next() { + var a domain.Approval + var subjectStr, tokenHash, decidedBy string + var payload []byte + if err := rows.Scan(&a.EntityID, &subjectStr, &a.Action, &a.RiskClass, &a.Kind, + &payload, &a.Status, &tokenHash, &a.ExpiresAt, &a.DecidedAt, + &decidedBy, &a.CreatedAt); err != nil { + return nil, err + } + a.TokenHash = tokenHash + if subjectStr != "" { + a.SubjectEntityID = domain.UUID(subjectStr) + } + if decidedBy != "" { + a.DecidedBy = domain.UUID(decidedBy) + } + if len(payload) > 0 { + var m map[string]any + if json.Unmarshal(payload, &m) == nil { + a.Payload = m + } + } + out = append(out, a) + } + return out, rows.Err() +} + +// Decide runs the full decide transaction. Returns the updated approval and, +// on approve with a linked execution, the Resume info for the driving adapter +// to dispatch the SSH work after commit. +func (r *ApprovalRepo) Decide(ctx context.Context, in ports.ApprovalDecideInput) (ports.ApprovalDecideResult, error) { + id := mustUUID(in.ApprovalID) + + tx, err := r.pool.Begin(ctx) + if err != nil { + return ports.ApprovalDecideResult{}, err + } + defer tx.Rollback(ctx) + + q := sqlcgen.New(tx) + + // Verify HMAC token if provided (single-use, S5). + if in.Token != "" { + var tokenHash *string + var apprStatus string + var expiresAt time.Time + err := tx.QueryRow(ctx, + "SELECT token_hash, status, expires_at FROM approvals WHERE entity_id = $1", + id).Scan(&tokenHash, &apprStatus, &expiresAt) + if err != nil || tokenHash == nil { + return ports.ApprovalDecideResult{}, domain.ErrNotFound + } + if apprStatus != "pending" { + return ports.ApprovalDecideResult{}, domain.ErrInvalidTransition + } + if expiresAt.Before(time.Now()) { + return ports.ApprovalDecideResult{}, domain.ErrInvalidTransition + } + if *tokenHash != hashApprovalToken(in.Token) { + return ports.ApprovalDecideResult{}, domain.ErrInvalidInput + } + } + + if err := q.UpdateApprovalStatus(ctx, sqlcgen.UpdateApprovalStatusParams{ + EntityID: id, + Status: in.Status, + }); err != nil { + return ports.ApprovalDecideResult{}, domain.ErrNotFound + } + + // Re-read approval. + app, err := q.GetApprovalByID(ctx, id) + if err != nil { + return ports.ApprovalDecideResult{}, err + } + approval := domain.Approval{ + EntityID: domain.UUID(app.EntityID.String()), + Action: app.Action, + RiskClass: app.RiskClass, + Kind: app.Kind, + Status: app.Status, + TokenHash: stringPtrOr(app.TokenHash), + ExpiresAt: app.ExpiresAt, + DecidedAt: app.DecidedAt, + DecidedBy: uuidPtrOr(app.DecidedBy), + CreatedAt: app.CreatedAt, + } + if app.SubjectEntityID != nil { + approval.SubjectEntityID = domain.UUID(app.SubjectEntityID.String()) + } + + // Audit + event (in-tx; NOTIFY fires post-commit). An audit failure must + // abort the decision — leaving no trace of who decided would violate the + // append-only audit contract (ADR 0016 §3.6). + actorLabel := in.Actor + if actorLabel == "" { + actorLabel = "unknown" + } + if auditErr := observability.Audit(ctx, q, "api", actorLabel, "decide", + &id, "POST", "/api/v1/approvals/"+id.String()+"/decision", "", + nil, + map[string]any{"decision": in.Status}); auditErr != nil { + return ports.ApprovalDecideResult{}, auditErr + } + if evErr := observability.Event(ctx, q, "approval.decided", &id, "info", "api", "", + map[string]any{"decision": in.Status, "actor": actorLabel}); evErr != nil { + return ports.ApprovalDecideResult{}, evErr + } + + result := ports.ApprovalDecideResult{Approval: approval} + + // Approve: un-gate the linked execution and open windows. + if in.Status == "approved" { + resume, aerr := r.approveSideEffects(ctx, tx, id) + if aerr != nil { + return result, aerr + } + result.Resume = resume + } else { + // Denied/revoked: reflect on the linked execution. + _, _ = tx.Exec(ctx, `UPDATE executions SET status = $2, completed_at = now() WHERE approval_id = $1 AND status = 'pending_approval'`, id, in.Status) + } + + // Flip any nomos session out of awaiting_input — the operator answered. + flipAwaitingSession(ctx, tx, id, in.Status) + + if err := tx.Commit(ctx); err != nil { + return result, err + } + return result, nil +} + +// approveSideEffects finds the linked pending execution, marks it approved, +// opens the assent window (+ destructive window for destructive risk), and +// returns the Resume info for the adapter to dispatch the SSH work. +func (r *ApprovalRepo) approveSideEffects(ctx context.Context, tx pgx.Tx, approvalID uuid.UUID) (*ports.ApprovalResume, error) { + var execID, targetID uuid.UUID + var actionStr, targetSlug, riskClass, sessionID string + err := tx.QueryRow(ctx, ` + SELECT e.entity_id, e.target_entity_id, e.action, e.risk_class, + COALESCE((SELECT pe.session_id FROM nomos_plan_executions pe + WHERE pe.execution_id = e.entity_id LIMIT 1)::text, '') + FROM executions e + WHERE e.approval_id = $1 AND e.status = 'pending_approval' + LIMIT 1`, approvalID).Scan(&execID, &targetID, &actionStr, &riskClass, &sessionID) + if err != nil { + slog.Warn("postgres: no pending execution found for approval", "approval_id", approvalID, "error", err) + return nil, nil + } + + _ = tx.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", targetID).Scan(&targetSlug) + + // Status only — risk_class was set correctly at request time (e.g. by + // policy.ClassifyCommand for `run`); overwriting it to a hardcoded + // 'config_mutation' here corrupted the audit ledger for every other risk + // class, including destructive. + _, _ = tx.Exec(ctx, `UPDATE executions SET status = 'approved' WHERE entity_id = $1`, execID) + + var agentID *uuid.UUID + if qerr := tx.QueryRow(ctx, "SELECT agent_id FROM executions WHERE entity_id = $1", execID).Scan(&agentID); qerr == nil && agentID != nil { + // Approving a plan step — by ANY route (the UI Approve button and + // chat-assent) — opens/extends the agent's assent window. The key is + // session-scoped so the window only auto-runs THIS task's plan steps + // (governance.go AssentWindowActive + session store's AssentWindowKey + // use the same "assent_window.agent:.session:" format). + expires := time.Now().Add(assentWindowDuration).UTC().Format(time.RFC3339) + _, _ = tx.Exec(ctx, `INSERT INTO autonomy_settings (key, value) VALUES ($1, $2) + ON CONFLICT (key) DO UPDATE SET value = $2`, + "assent_window.agent:"+agentID.String()+".session:"+sessionID, expires) + + // Approving a DESTRUCTIVE step via the button is exactly as explicit + // as a typed "I confirm" — open the same short, target+session-scoped + // destructive window chat-assent's typed-confirm path opens. + if riskClass == "destructive" && targetSlug != "" { + dExpires := time.Now().Add(destructiveWindowDuration).UTC().Format(time.RFC3339) + _, _ = tx.Exec(ctx, `INSERT INTO autonomy_settings (key, value) VALUES ($1, $2) + ON CONFLICT (key) DO UPDATE SET value = $2`, + "destructive_window.agent:"+agentID.String()+".target:"+targetSlug+".session:"+sessionID, dExpires) + } + } + + slog.Info("postgres: approved execution queued", + "execution_id", execID, "target", targetSlug, "action", actionStr) + return &ports.ApprovalResume{ + ExecutionID: domain.UUID(execID.String()), + TargetSlug: targetSlug, + Action: actionStr, + }, nil +} + +// flipAwaitingSession marks any nomos session linked to this approval as +// executing again (awaiting_input → executing), mirroring +// classifyAndGate's awaiting_input flip in reverse. +func flipAwaitingSession(ctx context.Context, tx pgx.Tx, approvalID uuid.UUID, status string) { + var awaitingSessionID string + _ = tx.QueryRow(ctx, ` + SELECT pe.session_id FROM nomos_plan_executions pe + JOIN executions ex ON ex.entity_id = pe.execution_id + WHERE ex.approval_id = $1 + LIMIT 1`, approvalID).Scan(&awaitingSessionID) + if awaitingSessionID == "" { + return + } + rtag, rerr := tx.Exec(ctx, ` + UPDATE agent_sessions SET status = 'executing', last_active_at = now() + WHERE id = $1 AND status = 'awaiting_input'`, awaitingSessionID) + if rerr != nil || rtag.RowsAffected() == 0 { + return + } + var taskEntID *uuid.UUID + var e uuid.UUID + if qerr := tx.QueryRow(ctx, `SELECT entity_id FROM agent_sessions WHERE id = $1`, awaitingSessionID).Scan(&e); qerr == nil && e != uuid.Nil { + taskEntID = &e + } + _ = observability.Event(ctx, sqlcgen.New(tx), "task.status", taskEntID, "info", "api", awaitingSessionID, + map[string]any{"status": "executing", "reason": "approval_decided", "decision": status}) +} + +// hashApprovalToken computes a SHA-256 hex-encoded hash of the token, used +// to verify single-use approval tokens (S5) without storing the plaintext. +func hashApprovalToken(token string) string { + h := sha256.Sum256([]byte(token)) + return hex.EncodeToString(h[:]) +} + +func uuidOrNil(id domain.UUID) any { + if id == "" { + return nil + } + return mustUUID(id) +} + +func stringPtrOr(s *string) string { + if s == nil { + return "" + } + return *s +} + +func uuidPtrOr(u *uuid.UUID) domain.UUID { + if u == nil { + return "" + } + return domain.UUID(u.String()) +} diff --git a/internal/execlog/execlog.go b/internal/adapters/postgres/execlog.go similarity index 69% rename from internal/execlog/execlog.go rename to internal/adapters/postgres/execlog.go index aab8b62b..8fef5c99 100644 --- a/internal/execlog/execlog.go +++ b/internal/adapters/postgres/execlog.go @@ -1,12 +1,4 @@ -// Package execlog persists incremental command output for an execution and -// announces it on the event stream. -// -// It exists as its own package because both SSH execution paths need it — -// internal/mcp (the agent's auto-run windows) and internal/httpapi (the -// post-approval actuator). Those two already carry near-identical copies of -// sshExec, and every bug found in this area so far has been a case of the two -// copies drifting apart; one shared sink is the cheap way not to repeat that. -package execlog +package db import ( "context" @@ -14,12 +6,15 @@ import ( "sync" "time" - "github.com/dtoro/oikos/internal/adapters/postgres" "github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen" "github.com/dtoro/oikos/internal/observability" "github.com/google/uuid" ) +// Execution log persistence + throttled event emission (folded from the +// internal/execlog package during the hex refactor — the execution-log +// repository lives with the other execution persistence in this adapter). + // eventInterval throttles execution.output events. Chunks are persisted as // they arrive, but a chatty command (apt, a long build) can produce hundreds // per second and the SSE broker drops events for slow subscribers — flooding @@ -28,16 +23,16 @@ import ( // the rows. const eventInterval = time.Second -// Sink receives output chunks as they arrive from a remote command. -type Sink func(stream string, chunk []byte) +// ExecLogSink receives output chunks as they arrive from a remote command. +type ExecLogSink func(stream string, chunk []byte) -// New returns a Sink that writes chunks to execution_logs and emits a -// throttled execution.output event, plus a Flush to call when the command -// finishes. +// NewExecutionLog returns a Sink that writes chunks to execution_logs and +// emits a throttled execution.output event, plus a Flush to call when the +// command finishes. // // The returned Sink is safe for concurrent use: stdout and stderr are written // from separate goroutines. -func New(ctx context.Context, pool *db.Pool, execID uuid.UUID, correlationID string) (Sink, func()) { +func NewExecutionLog(ctx context.Context, pool *Pool, execID uuid.UUID, correlationID string) (ExecLogSink, func()) { var ( mu sync.Mutex seq int @@ -102,8 +97,16 @@ func New(ctx context.Context, pool *db.Pool, execID uuid.UUID, correlationID str return sink, flush } -// Read returns an execution's persisted output in order. -func Read(ctx context.Context, pool *db.Pool, execID uuid.UUID, limit int) ([]Chunk, error) { +// ExecLogChunk is one persisted slice of command output. +type ExecLogChunk struct { + Seq int `json:"seq"` + Stream string `json:"stream"` + Chunk string `json:"chunk"` + TS time.Time `json:"ts"` +} + +// ReadExecutionLog returns an execution's persisted output in order. +func ReadExecutionLog(ctx context.Context, pool *Pool, execID uuid.UUID, limit int) ([]ExecLogChunk, error) { if limit <= 0 { limit = 1000 } @@ -115,9 +118,9 @@ func Read(ctx context.Context, pool *db.Pool, execID uuid.UUID, limit int) ([]Ch } defer rows.Close() - var out []Chunk + var out []ExecLogChunk for rows.Next() { - var c Chunk + var c ExecLogChunk if err := rows.Scan(&c.Seq, &c.Stream, &c.Chunk, &c.TS); err != nil { return nil, err } @@ -125,11 +128,3 @@ func Read(ctx context.Context, pool *db.Pool, execID uuid.UUID, limit int) ([]Ch } return out, rows.Err() } - -// Chunk is one persisted slice of command output. -type Chunk struct { - Seq int `json:"seq"` - Stream string `json:"stream"` - Chunk string `json:"chunk"` - TS time.Time `json:"ts"` -} diff --git a/internal/adapters/postgres/governance.go b/internal/adapters/postgres/governance.go index 4966422b..de8a0763 100644 --- a/internal/adapters/postgres/governance.go +++ b/internal/adapters/postgres/governance.go @@ -234,15 +234,32 @@ func (r *ExecRunRepo) MarkRunning(ctx context.Context, id domain.UUID) error { } // Finalize stamps the terminal status, result payload, duration, and -// completion time. +// completion time, and emits an execution.{status} event correlated to the +// session (the correlation_id is read from the row so every execution path +// — MCP auto-run, approval, worker — gets a correlated event). func (r *ExecRunRepo) Finalize(ctx context.Context, in ports.FinalizeExecutionInput) error { + id := mustUUID(in.ExecutionID) _, err := r.pool.Exec(ctx, `UPDATE executions SET status=$2, result=$3::jsonb, duration_ms=$4, started_at=$5, completed_at=now() WHERE entity_id=$1`, - mustUUID(in.ExecutionID), in.Status, string(in.Result), + id, in.Status, string(in.Result), int(time.Since(in.StartedAt).Milliseconds()), in.StartedAt) - return err + if err != nil { + return err + } + // Emit a correlated execution event for SSE fan-out. The correlation_id + // was set on the row at creation time (CreateRun, InsertExecution, etc.) + // and is preserved through the lifecycle. + var correlationID string + _ = r.pool.QueryRow(ctx, `SELECT correlation_id FROM executions WHERE entity_id = $1`, id).Scan(&correlationID) + severity := "info" + if in.Status == "failed" { + severity = "warning" + } + _ = observability.Event(ctx, sqlcgen.New(r.pool), "execution."+in.Status, &id, severity, "execution", correlationID, + map[string]any{"execution_id": in.ExecutionID, "status": in.Status}) + return nil } // QueueApproval creates the approval row, flips the execution to diff --git a/internal/core/app/approval.go b/internal/core/app/approval.go new file mode 100644 index 00000000..aa0059be --- /dev/null +++ b/internal/core/app/approval.go @@ -0,0 +1,65 @@ +package app + +import ( + "context" + "fmt" + + "github.com/dtoro/oikos/internal/core/domain" + "github.com/dtoro/oikos/internal/core/ports" + "github.com/google/uuid" +) + +// ApprovalService is the approval-decision use-case: list pending approvals +// and decide (approve/deny/revoke), which un-gates the linked execution and +// persists the audit trail. The driving adapter dispatches the SSH work for +// approved executions after Decide returns. +type ApprovalService struct { + approvals ports.ApprovalRepository +} + +// NewApprovalService wires the service. +func NewApprovalService(approvals ports.ApprovalRepository) *ApprovalService { + return &ApprovalService{approvals: approvals} +} + +// List returns pending approvals for the given entity. +func (s *ApprovalService) List(ctx context.Context, entityID domain.UUID, limit int) ([]domain.Approval, error) { + return s.approvals.ListPending(ctx, entityID, limit) +} + +// ApprovalDecideCmd is one decide-approval request. +type ApprovalDecideCmd struct { + ApprovalID domain.UUID + Token string + Decision string // "approve" | "deny" | "revoke" + Actor string +} + +// Decide validates the decision, maps to status, and delegates the +// transactional flip + execution un-gate + audit + event to the repository. +// Returns the updated approval and, on approve, the execution to resume. +func (s *ApprovalService) Decide(ctx context.Context, cmd ApprovalDecideCmd) (ports.ApprovalDecideResult, error) { + var status string + switch cmd.Decision { + case "approve": + status = "approved" + case "deny": + status = "denied" + case "revoke": + status = "revoked" + default: + return ports.ApprovalDecideResult{}, fmt.Errorf("%w: invalid decision %q", domain.ErrInvalidInput, cmd.Decision) + } + + aid, err := uuid.Parse(string(cmd.ApprovalID)) + if err != nil { + return ports.ApprovalDecideResult{}, fmt.Errorf("parse approval id: %w", err) + } + + return s.approvals.Decide(ctx, ports.ApprovalDecideInput{ + ApprovalID: domain.UUID(aid.String()), + Token: cmd.Token, + Status: status, + Actor: cmd.Actor, + }) +} diff --git a/internal/core/app/approval_test.go b/internal/core/app/approval_test.go new file mode 100644 index 00000000..cbf69c8d --- /dev/null +++ b/internal/core/app/approval_test.go @@ -0,0 +1,114 @@ +package app + +import ( + "context" + "errors" + "testing" + + "github.com/dtoro/oikos/internal/core/domain" + "github.com/dtoro/oikos/internal/core/ports" + "github.com/dtoro/oikos/internal/core/ports/portstest" + "github.com/google/uuid" +) + +func TestApprovalDecideValidDecisions(t *testing.T) { + for _, tc := range []struct { + decision string + want string + }{ + {"approve", "approved"}, + {"deny", "denied"}, + {"revoke", "revoked"}, + } { + t.Run(tc.decision, func(t *testing.T) { + repo := portstest.NewApprovalRepo() + repo.Result = ports.ApprovalDecideResult{ + Approval: domain.Approval{EntityID: "some-id", Status: tc.want}, + } + svc := NewApprovalService(repo) + + aid := uuid.New() + got, err := svc.Decide(context.Background(), ApprovalDecideCmd{ + ApprovalID: domain.UUID(aid.String()), + Decision: tc.decision, + Actor: "operator", + }) + if err != nil { + t.Fatalf("Decide(%s): %v", tc.decision, err) + } + if len(repo.Decided) != 1 { + t.Fatalf("Decide called %d times, want 1", len(repo.Decided)) + } + in := repo.Decided[0] + if in.Status != tc.want { + t.Errorf("repo status = %q, want %q", in.Status, tc.want) + } + if in.Actor != "operator" || string(in.ApprovalID) != aid.String() { + t.Errorf("repo input = %+v", in) + } + if got.Approval.Status != tc.want { + t.Errorf("result approval status = %q, want %q", got.Approval.Status, tc.want) + } + }) + } +} + +func TestApprovalDecideInvalidDecision(t *testing.T) { + repo := portstest.NewApprovalRepo() + svc := NewApprovalService(repo) + _, err := svc.Decide(context.Background(), ApprovalDecideCmd{ + ApprovalID: domain.UUID(uuid.New().String()), + Decision: "maybe", + Actor: "operator", + }) + if !errors.Is(err, domain.ErrInvalidInput) { + t.Fatalf("err = %v, want ErrInvalidInput", err) + } + if len(repo.Decided) != 0 { + t.Error("repo must not be called on an invalid decision") + } +} + +func TestApprovalDecideRepoErrorPropagates(t *testing.T) { + repo := portstest.NewApprovalRepo() + repo.ErrStub = domain.ErrNotFound + svc := NewApprovalService(repo) + _, err := svc.Decide(context.Background(), ApprovalDecideCmd{ + ApprovalID: domain.UUID(uuid.New().String()), + Decision: "approve", + }) + if !errors.Is(err, domain.ErrNotFound) { + t.Fatalf("err = %v, want ErrNotFound", err) + } +} + +func TestApprovalList(t *testing.T) { + repo := portstest.NewApprovalRepo() + repo.Pending = []domain.Approval{{EntityID: "a", Status: "pending"}} + svc := NewApprovalService(repo) + got, err := svc.List(context.Background(), "e1", 10) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(got) != 1 || got[0].Status != "pending" { + t.Fatalf("got %+v", got) + } +} + +func TestApprovalDecideTokenForwarded(t *testing.T) { + repo := portstest.NewApprovalRepo() + svc := NewApprovalService(repo) + aid := uuid.New() + _, err := svc.Decide(context.Background(), ApprovalDecideCmd{ + ApprovalID: domain.UUID(aid.String()), + Token: "hmac-token", + Decision: "approve", + Actor: "operator", + }) + if err != nil { + t.Fatalf("Decide: %v", err) + } + if repo.Decided[0].Token != "hmac-token" { + t.Errorf("token = %q, want forwarded", repo.Decided[0].Token) + } +} diff --git a/internal/core/app/execution.go b/internal/core/app/execution.go index 7f707fbd..32176cf9 100644 --- a/internal/core/app/execution.go +++ b/internal/core/app/execution.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "log/slog" + "strings" "time" "github.com/dtoro/oikos/internal/core/domain" @@ -19,10 +20,10 @@ import ( // (MCP `run`/`docker_exec` tools, and any future REST surface) converge // here — one policy, one audit trail (plan §4.A). type ExecutionService struct { - policy *PolicyService - exec ports.CommandExecutor - resolver ports.TargetResolver - recorder ports.ExecutionRecorder + policy *PolicyService + exec ports.CommandExecutor + resolver ports.TargetResolver + recorder ports.ExecutionRecorder } // NewExecutionService wires the service. @@ -199,6 +200,31 @@ func (s *ExecutionService) finalize(ctx context.Context, execID domain.UUID, sta } } +// DispatchQueued runs a previously-recorded queued execution (status +// 'proposed' or 'approved') over the CommandExecutor port. The command is +// derived from the execution's action column the same way the legacy worker +// did: "action_name:{json_params}" extracts params["command"], else the +// action string is the raw command. Used by the execworker poller adapter. +func (s *ExecutionService) DispatchQueued(ctx context.Context, execID domain.UUID, targetSlug, action string) (string, error) { + return s.dispatch(ctx, execID, targetSlug, queuedCommand(action), nil, nil) +} + +// queuedCommand extracts the actual shell command from an execution's action +// column. Format: "action_name:{json_params}" → params["command"]; otherwise +// the action string is itself the raw command. +func queuedCommand(action string) string { + if idx := strings.Index(action, ":"); idx > 0 && idx < len(action)-1 { + rawParams := action[idx+1:] + var params map[string]any + if json.Unmarshal([]byte(rawParams), ¶ms) == nil { + if c, ok := params["command"].(string); ok && c != "" { + return c + } + } + } + return action +} + func jsonOutBytes(out string) []byte { b, _ := json.Marshal(map[string]any{"output": out}) return b diff --git a/internal/core/app/execution_test.go b/internal/core/app/execution_test.go index 42d4f56f..13c916ff 100644 --- a/internal/core/app/execution_test.go +++ b/internal/core/app/execution_test.go @@ -281,3 +281,135 @@ func TestExecutionSubmitHostHint(t *testing.T) { t.Errorf("default hint missing: %s", d.Message) } } + +func TestQueuedCommandExtraction(t *testing.T) { + cases := map[string]string{ + `run:{"command":"df -h","purpose":"check"}`: "df -h", + `systemctl restart caddy`: "systemctl restart caddy", + `pct:{"command":"pct list"}`: "pct list", + `apt_upgrade:not-json`: "apt_upgrade:not-json", + `plain command`: "plain command", + } + for in, want := range cases { + if got := queuedCommand(in); got != want { + t.Errorf("queuedCommand(%q) = %q, want %q", in, got, want) + } + } +} + +func TestExecutionDispatchQueued(t *testing.T) { + rec, exec, _ := newExecDeps(t) + exec.Results = []ports.ExecResult{{Output: "uptime output"}} + + res, err := svcFor(t, rec, exec).DispatchQueued(context.Background(), targetID, "lxc:x", `run:{"command":"uptime"}`) + if err != nil { + t.Fatalf("DispatchQueued: %v", err) + } + if res != "uptime output" { + t.Errorf("result = %q", res) + } + if len(rec.Running) != 1 || len(rec.Finalized) != 1 { + t.Fatalf("running=%d finalized=%d", len(rec.Running), len(rec.Finalized)) + } + if rec.Finalized[0].Status != "completed" { + t.Errorf("finalized = %+v", rec.Finalized[0]) + } +} + +func TestExecutionDispatchQueuedFailure(t *testing.T) { + rec, exec, _ := newExecDeps(t) + exec.Results = []ports.ExecResult{{Output: "boom", Err: errors.New("exit 2")}} + + _, err := svcFor(t, rec, exec).DispatchQueued(context.Background(), targetID, "lxc:x", "uptime") + if err == nil { + t.Fatal("expected dispatch error") + } + if rec.Finalized[0].Status != "failed" { + t.Errorf("finalized = %+v, want failed", rec.Finalized[0]) + } +} + +func TestExecutionDispatchQueuedResolverFailure(t *testing.T) { + rec, exec, _ := newExecDeps(t) + failing := &portstest.FakeResolver{Err: errors.New("no route")} + svc := NewExecutionService(NewPolicyService(portstest.NewGovernanceStore()), exec, failing, rec) + + _, err := svc.DispatchQueued(context.Background(), targetID, "lxc:x", "uptime") + if err == nil { + t.Fatal("expected resolver error") + } + if len(exec.Calls) != 0 { + t.Error("no execution on resolve failure") + } + if rec.Finalized[0].Status != "failed" { + t.Errorf("finalized = %+v, want failed", rec.Finalized[0]) + } +} + +func TestExecutionSubmitNoSessionGetsCorrelationID(t *testing.T) { + rec, exec, svc := newExecDeps(t) + exec.Results = []ports.ExecResult{{Output: "ok"}} + + res := svc.Submit(context.Background(), ExecutionSubmitCmd{ + AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x", + Command: "df -h", // auto-run, no session + }) + if res.Err != nil { + t.Fatalf("submit: %v", res.Err) + } + if len(rec.Created) != 1 || rec.Created[0].CorrelationID == "" { + t.Errorf("created = %+v, want a correlation id", rec.Created) + } +} + +func TestExecutionSubmitAsyncStarted(t *testing.T) { + rec, exec, _ := newExecDeps(t) + exec.Results = []ports.ExecResult{{Output: "sleeping"}} + store := portstest.NewGovernanceStore() + store.PlanSessions[sessionOK] = true + store.Assent[string(agentID)+"/"+sessionOK] = true + svc := NewExecutionService(NewPolicyService(store), exec, &portstest.FakeResolver{Addr: "10.0.0.1", User: "root"}, rec) + + res := svc.Submit(context.Background(), ExecutionSubmitCmd{ + AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x", + Command: "sleep 60", SessionID: sessionOK, Async: true, + }) + if !res.AsyncStarted { + t.Fatalf("result = %+v, want async started", res) + } + if res.ExecutionID == "" { + t.Fatal("async submission must return the execution id synchronously") + } +} + +func TestExecutionSubmitAutoRunQueueResume(t *testing.T) { + rec, exec, _ := newExecDeps(t) + exec.Results = []ports.ExecResult{{Output: "done"}} + + // A gated command queues; the same command with a live assent window + // auto-runs instead — the window route switch in Submit. + store := portstest.NewGovernanceStore() + store.PlanSessions[sessionOK] = true + store.Assent[string(agentID)+"/"+sessionOK] = true + svc := NewExecutionService(NewPolicyService(store), exec, &portstest.FakeResolver{Addr: "10.0.0.1", User: "root"}, rec) + + res := svc.Submit(context.Background(), ExecutionSubmitCmd{ + AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x", + Command: "systemctl enable caddy", SessionID: sessionOK, + }) + if res.Decision.Action != DecisionAuto { + t.Fatalf("decision = %+v, want auto (via assent window)", res.Decision) + } + if len(rec.Queued) != 0 { + t.Errorf("window route must not queue, got %+v", rec.Queued) + } +} + +// svcFor builds a service over the given recorder/executor with a plan-backed +// governance store. +func svcFor(t *testing.T, rec *portstest.ExecutionRecorder, exec *portstest.RecordingExecutor) *ExecutionService { + t.Helper() + store := portstest.NewGovernanceStore() + store.PlanSessions[sessionOK] = true + return NewExecutionService(NewPolicyService(store), exec, &portstest.FakeResolver{Addr: "10.1.1.1", User: "root"}, rec) +} diff --git a/internal/core/ports/governance.go b/internal/core/ports/governance.go index bd458b51..71ef061d 100644 --- a/internal/core/ports/governance.go +++ b/internal/core/ports/governance.go @@ -83,17 +83,32 @@ type ExecutionRepository interface { type ApprovalDecideInput struct { ApprovalID domain.UUID Token string - Approved bool + Status string // "approved" | "denied" | "revoked" Actor string Audit []AuditEntry Event *Event } +// ApprovalResume identifies an un-gated execution the driving adapter should +// dispatch after an approved decision (SSH work is adapter-specific). +type ApprovalResume struct { + ExecutionID domain.UUID + TargetSlug string + Action string +} + +// ApprovalDecideResult carries the updated approval and, on approve, the +// linked execution to resume. +type ApprovalDecideResult struct { + Approval domain.Approval + Resume *ApprovalResume +} + // ApprovalRepository is the approval aggregate. type ApprovalRepository interface { ListPending(ctx context.Context, entityID domain.UUID, limit int) ([]domain.Approval, error) - Decide(ctx context.Context, input ApprovalDecideInput) (domain.Approval, error) + Decide(ctx context.Context, input ApprovalDecideInput) (ApprovalDecideResult, error) } // GovernanceStore is the policy read model: the facts the gating decision diff --git a/internal/core/ports/portstest/fakes.go b/internal/core/ports/portstest/fakes.go index 265d3d5e..990c10d9 100644 --- a/internal/core/ports/portstest/fakes.go +++ b/internal/core/ports/portstest/fakes.go @@ -17,11 +17,11 @@ import ( // EntityRepo is an in-memory ports.EntityRepository. Command inputs' audit, // event, and derived-check fields are recorded for assertion. type EntityRepo struct { - mu sync.Mutex - byID map[domain.UUID]domain.Entity - bySlug map[string]domain.UUID - order []domain.UUID - nextID int + mu sync.Mutex + byID map[domain.UUID]domain.Entity + bySlug map[string]domain.UUID + order []domain.UUID + nextID int Audits []ports.AuditEntry Events []ports.Event Checks map[domain.UUID][]ports.CheckDef @@ -34,8 +34,8 @@ type EntityRepo struct { // NewEntityRepo builds an empty in-memory entity repository. func NewEntityRepo() *EntityRepo { return &EntityRepo{ - byID: make(map[domain.UUID]domain.Entity), - bySlug: make(map[string]domain.UUID), + byID: make(map[domain.UUID]domain.Entity), + bySlug: make(map[string]domain.UUID), Checks: make(map[domain.UUID][]ports.CheckDef), Idempotent: make(map[string]ports.IdempotentResponse), } @@ -404,12 +404,12 @@ func (r *FakeResolver) IsGuest(entityType string) bool { // FakeProvisioner records provision requests and replies with canned // results (default: a successful create echoing the request's VMID). type FakeProvisioner struct { - mu sync.Mutex - LXCs []ports.LXCInput - VMs []ports.VMInput - LXCRes ports.ProvisionResult - LXCErr error - VMErr error + mu sync.Mutex + LXCs []ports.LXCInput + VMs []ports.VMInput + LXCRes ports.ProvisionResult + LXCErr error + VMErr error } // CreateLXC records the request and replies with the canned result. @@ -623,7 +623,42 @@ func (r *ExecutionRecorder) QueueApproval(_ context.Context, in ports.QueueAppro // MuLock exposes the internal mutex for tests that need to inspect state // racing the async dispatch goroutine. -func (r *ExecutionRecorder) MuLock() { r.mu.Lock() } +func (r *ExecutionRecorder) MuLock() { r.mu.Lock() } // MuUnlock releases the internal mutex. func (r *ExecutionRecorder) MuUnlock() { r.mu.Unlock() } + +// ApprovalRepo is an in-memory ports.ApprovalRepository. ListPending returns +// the canned pending list; Decide records the input and replies with a canned +// result or the ErrStub. +type ApprovalRepo struct { + mu sync.Mutex + // Pending is returned by ListPending. + Pending []domain.Approval + // Decided records every Decide input for assertion. + Decided []ports.ApprovalDecideInput + // Result is returned by Decide (default: empty approval). + Result ports.ApprovalDecideResult + ErrStub error +} + +// NewApprovalRepo builds an empty in-memory approval repository. +func NewApprovalRepo() *ApprovalRepo { return &ApprovalRepo{} } + +// ListPending returns the canned pending list. +func (r *ApprovalRepo) ListPending(_ context.Context, _ domain.UUID, _ int) ([]domain.Approval, error) { + r.mu.Lock() + defer r.mu.Unlock() + return r.Pending, r.ErrStub +} + +// Decide records the input and replies with the canned result. +func (r *ApprovalRepo) Decide(_ context.Context, in ports.ApprovalDecideInput) (ports.ApprovalDecideResult, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.Decided = append(r.Decided, in) + if r.ErrStub != nil { + return ports.ApprovalDecideResult{}, r.ErrStub + } + return r.Result, nil +} diff --git a/internal/execworker/worker.go b/internal/execworker/worker.go index 84ea646e..ad45cbd8 100644 --- a/internal/execworker/worker.go +++ b/internal/execworker/worker.go @@ -1,23 +1,25 @@ // Package execworker processes pending executions as a background daemon. // This provides a Postgres-backed queue: executions survive restarts, and // per-execution advisory locks prevent duplicate processing across instances. +// +// Since Phase 4 of the hexagonal refactor, the dispatch logic lives in +// ExecutionService.DispatchQueued (over CommandExecutor + TargetResolver +// ports); this package is a thin poller adapter that claims queued rows and +// delegates to the service. package execworker import ( "context" - "encoding/json" - "fmt" "log/slog" - "os" - "strings" "time" - "github.com/dtoro/oikos/internal/actuator" - "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/adapters/postgres" - "github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen" + "github.com/dtoro/oikos/internal/adapters/remote" + "github.com/dtoro/oikos/internal/adapters/ssh" + "github.com/dtoro/oikos/internal/config" + "github.com/dtoro/oikos/internal/core/app" + "github.com/dtoro/oikos/internal/core/domain" "github.com/dtoro/oikos/internal/health" - "github.com/dtoro/oikos/internal/remote" "github.com/google/uuid" ) @@ -25,7 +27,12 @@ import ( func Run(ctx context.Context, pool *db.Pool, cfg config.Config) { slog.Info("execworker: starting") - // Liveness probe + // Build the ExecutionService with the same wiring as cmd/oikos/main.go. + executor := ssh.NewExecutor(ssh.FileSignerSource(), 5*time.Minute) + resolver := remote.NewResolver(pool) + policySvc := app.NewPolicyService(db.NewGovernanceRepo(pool)) + execSvc := app.NewExecutionService(policySvc, executor, resolver, db.NewExecRunRepo(pool)) + probe := health.New(2 * time.Minute) probe.Serve(ctx, cfg.HealthListen) @@ -41,7 +48,7 @@ func Run(ctx context.Context, pool *db.Pool, cfg config.Config) { slog.Info("execworker: shutting down") return case <-ticker.C: - processPending(ctx, pool) + processPending(ctx, pool, execSvc) probe.Bump() } } @@ -59,11 +66,11 @@ func recoverOrphaned(ctx context.Context, pool *db.Pool) { } } -// processPending polls for pending executions and dispatches them. -func processPending(ctx context.Context, pool *db.Pool) { +// processPending polls for pending executions and dispatches them via +// ExecutionService.DispatchQueued. +func processPending(ctx context.Context, pool *db.Pool, execSvc *app.ExecutionService) { rows, err := pool.Query(ctx, ` - SELECT e.entity_id, e.target_entity_id, e.action, e.risk_class, e.correlation_id, e.status, - COALESCE(t.slug, '') AS target_slug + SELECT e.entity_id, COALESCE(t.slug, '') AS target_slug, e.action FROM executions e LEFT JOIN entities t ON t.id = e.target_entity_id WHERE e.status = 'proposed' @@ -75,23 +82,16 @@ func processPending(ctx context.Context, pool *db.Pool) { } defer rows.Close() - q := sqlcgen.New(pool) - for rows.Next() { - var execID, targetID *uuid.UUID - var action, riskClass, correlationID, status, targetSlug string - if err := rows.Scan(&execID, &targetID, &action, &riskClass, &correlationID, &status, &targetSlug); err != nil { + var execID uuid.UUID + var targetSlug, action string + if err := rows.Scan(&execID, &targetSlug, &action); err != nil { slog.Error("execworker: scan row", "error", err) continue } - if execID == nil { - continue - } // At-most-once: try advisory lock on execution entity_id. - // Acquire a dedicated connection so the session-scoped lock isn't - // released when the transient pool connection is returned. - lockKey := hashUUID(*execID) + lockKey := hashUUID(execID) lockConn, err := pool.Acquire(ctx) if err != nil { slog.Error("execworker: acquire lock conn", "error", err) @@ -103,99 +103,24 @@ func processPending(ctx context.Context, pool *db.Pool) { continue } - dispatch(ctx, pool, q, *execID, targetID, action, targetSlug, correlationID) + // Release the lock on the same connection even if the dispatch + // panics — an un-released advisory lock would permanently orphan the + // execution (every future worker skips it at pg_try_advisory_lock). + func() { + defer func() { + if r := recover(); r != nil { + slog.Error("execworker: panic dispatching execution", "execution", execID, "panic", r) + } + lockConn.Exec(context.WithoutCancel(ctx), "SELECT pg_advisory_unlock($1)", lockKey) + lockConn.Release() + }() - // Release the per-execution lock on the same connection. - lockConn.Exec(context.WithoutCancel(ctx), "SELECT pg_advisory_unlock($1)", lockKey) - lockConn.Release() - } -} - -func dispatch(ctx context.Context, pool *db.Pool, q *sqlcgen.Queries, execID uuid.UUID, targetID *uuid.UUID, action, targetSlug, correlationID string) { - startedAt := time.Now() - - // Mark running - _, err := pool.Exec(ctx, - `UPDATE executions SET status = 'running', started_at = $2 WHERE entity_id = $1`, - execID, startedAt) - if err != nil { - slog.Error("execworker: mark running", "error", err, "execution", execID) - return - } - - // Resolve SSH target. If targetSlug is available, use it; otherwise resolve from targetID. - var host, user string - if targetSlug == "" && targetID != nil { - if err := pool.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", *targetID).Scan(&targetSlug); err != nil { - failExecution(ctx, pool, execID, fmt.Sprintf("resolve target slug: %v", err)) - return - } - } - if targetSlug != "" { - addr, sshUser, err := remote.ResolveHost(ctx, pool, targetSlug, "root") - if err == nil { - host, user = addr, sshUser - } - } - if host == "" { - failExecution(ctx, pool, execID, fmt.Sprintf("no reachable target: %s", targetSlug)) - return - } - - // Determine the command to run from the action field. - // Format: "action_name:{json_params}" or a raw command string. - cmd := action - if idx := strings.Index(action, ":"); idx > 0 && idx < len(action)-1 { - rawParams := action[idx+1:] - var params map[string]any - if json.Unmarshal([]byte(rawParams), ¶ms) == nil { - if c, ok := params["command"].(string); ok && c != "" { - cmd = c + _, dispatchErr := execSvc.DispatchQueued(ctx, domain.UUID(execID.String()), targetSlug, action) + if dispatchErr != nil { + slog.Error("execworker: dispatch failed", "execution", execID, "error", dispatchErr) } - } + }() } - - signer, err := actuator.LoadSigner(os.Getenv("OIKOS_SSH_KEY_PATH")) - if err != nil { - signer, err = actuator.LoadSigner("/etc/oikos/ssh_key") - if err != nil { - failExecution(ctx, pool, execID, fmt.Sprintf("load ssh key: %v", err)) - return - } - } - - client, err := actuator.Dial(ctx, actuator.DialOptions{Host: host, User: user, Signer: signer}) - if err != nil { - failExecution(ctx, pool, execID, fmt.Sprintf("ssh dial: %v", err)) - return - } - defer client.Close() - - out, err := actuator.RunCombinedOutput(ctx, client, cmd) - if err != nil { - failExecution(ctx, pool, execID, fmt.Sprintf("command: %v\noutput: %s", err, string(out))) - return - } - - duration := time.Since(startedAt).Milliseconds() - resultJSON, _ := json.Marshal(map[string]any{"output": string(out), "success": true}) - _ = q.UpdateExecutionStatus(ctx, sqlcgen.UpdateExecutionStatusParams{ - EntityID: execID, - Status: "completed", - Result: resultJSON, - DurationMs: &[]int32{int32(duration)}[0], - Verified: true, - }) - - slog.Info("execworker: execution complete", - "execution", execID, "target", targetSlug, "duration_ms", duration) -} - -func failExecution(ctx context.Context, pool *db.Pool, execID uuid.UUID, reason string) { - slog.Error("execworker: execution failed", "execution", execID, "error", reason) - resultJSON, _ := json.Marshal(map[string]any{"error": reason, "success": false}) - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb, completed_at=now() WHERE entity_id=$1`, - execID, resultJSON) } func hashUUID(id uuid.UUID) int { @@ -204,4 +129,4 @@ func hashUUID(id uuid.UUID) int { h = (h*31 + int(b)) & 0x7fffffff } return h -} \ No newline at end of file +} diff --git a/internal/httpapi/actuator.go b/internal/httpapi/actuator.go index d223c99b..374bdae2 100644 --- a/internal/httpapi/actuator.go +++ b/internal/httpapi/actuator.go @@ -14,7 +14,6 @@ import ( "github.com/dtoro/oikos/internal/adapters/postgres" "github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen" "github.com/dtoro/oikos/internal/core/app" - "github.com/dtoro/oikos/internal/execlog" "github.com/dtoro/oikos/internal/observability" "github.com/google/uuid" ) @@ -80,7 +79,7 @@ const sshExecTimeout = 10 * time.Minute // sshExecStream runs a command and reports its combined output, forwarding // each chunk to sink as it arrives. A nil sink behaves exactly as before. -func sshExecStream(ctx context.Context, host, user, command string, sink execlog.Sink) (string, error) { +func sshExecStream(ctx context.Context, host, user, command string, sink db.ExecLogSink) (string, error) { initSSH() if len(_sshKey) == 0 { return "", fmt.Errorf("no SSH key available") @@ -260,7 +259,7 @@ func (s *Server) executeApprovedAction(ctx context.Context, pool *db.Pool, execI `SELECT correlation_id FROM executions WHERE entity_id = $1`, execID).Scan(&correlationID); qerr != nil { correlationID = "" } - sink, flushLogs := execlog.New(ctx, pool, execID, correlationID) + sink, flushLogs := db.NewExecutionLog(ctx, pool, execID, correlationID) defer flushLogs() var output, cmd string diff --git a/internal/httpapi/api_test.go b/internal/httpapi/api_test.go index 6706d697..b87db400 100644 --- a/internal/httpapi/api_test.go +++ b/internal/httpapi/api_test.go @@ -16,8 +16,8 @@ import ( "testing" "time" - "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/adapters/postgres" + "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/core/app" "github.com/dtoro/oikos/internal/core/ports/portstest" "github.com/jackc/pgx/v5" @@ -109,7 +109,8 @@ func newTestHandler(t *testing.T, cfg config.Config) http.Handler { app.NewPolicyService(portstest.NewGovernanceStore()), &portstest.RecordingExecutor{}, &portstest.FakeResolver{Addr: "127.0.0.1"}, portstest.NewExecutionRecorder()) - return NewHandler(handlerCtx, pool, cfg, app.NewEntityService(repo, onto), repo, db.NewEntityReader(pool), app.NewRelationshipService(db.NewRelRepo(pool), onto), provisioning, seeds, execSvc) + approvalSvc := app.NewApprovalService(db.NewApprovalRepo(pool)) + return NewHandler(handlerCtx, pool, cfg, app.NewEntityService(repo, onto), repo, db.NewEntityReader(pool), app.NewRelationshipService(db.NewRelRepo(pool), onto), provisioning, seeds, execSvc, approvalSvc) } // testAuthToken is the static bearer token devConfig() configures. There is diff --git a/internal/httpapi/approvals.go b/internal/httpapi/approvals.go index 78724daf..6b6aa018 100644 --- a/internal/httpapi/approvals.go +++ b/internal/httpapi/approvals.go @@ -5,15 +5,12 @@ import ( "encoding/json" "fmt" "log/slog" - "time" - "github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen" + "github.com/dtoro/oikos/internal/core/app" "github.com/dtoro/oikos/internal/core/domain" "github.com/dtoro/oikos/internal/httpapi/gen" - "github.com/dtoro/oikos/internal/observability" "github.com/dtoro/oikos/internal/safego" "github.com/google/uuid" - "github.com/jackc/pgx/v5" ) // ─── Approvals ───────────────────────────────────────────────────────── @@ -22,8 +19,8 @@ func (s *Server) ListApprovals(ctx context.Context, req gen.ListApprovalsRequest limit := clampLimit(req.Params.Limit) var status *string if req.Params.Status != nil { - s := string(*req.Params.Status) - status = &s + st := string(*req.Params.Status) + status = &st } var kind *string if req.Params.Kind != nil { @@ -79,6 +76,10 @@ func (s *Server) ListApprovals(ctx context.Context, req gen.ListApprovalsRequest return gen.ListApprovals200JSONResponse{Items: items, NextCursor: next}, nil } +// DecideApproval is the thin presenter over ApprovalService.Decide: it maps +// the request onto the service command, then dispatches the SSH work for an +// approved execution in a background goroutine. The HMAC/window/execution +// un-gating logic lives in the service + repository (Phase 9 convergence). func (s *Server) DecideApproval(ctx context.Context, req gen.DecideApprovalRequestObject) (gen.DecideApprovalResponseObject, error) { if req.Body == nil { return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput) @@ -89,184 +90,43 @@ func (s *Server) DecideApproval(ctx context.Context, req gen.DecideApprovalReque return nil, err } - actorType, actor := actorInfo(ctx) + _, actor := actorInfo(ctx) - tx, err := s.pool.Begin(ctx) - if err != nil { - return nil, err - } - defer tx.Rollback(ctx) - - q := sqlcgen.New(tx) - - // Verify HMAC token if provided (single-use, S5). - if req.Body.Token != nil && *req.Body.Token != "" { - var tokenHash *string - var apprStatus string - var expiresAt time.Time - err := tx.QueryRow(ctx, - "SELECT token_hash, status, expires_at FROM approvals WHERE entity_id = $1", - id).Scan(&tokenHash, &apprStatus, &expiresAt) - if err != nil || tokenHash == nil { - return nil, fmt.Errorf("%w: approval not found", domain.ErrNotFound) - } - if apprStatus != "pending" { - return nil, fmt.Errorf("%w: approval already decided", domain.ErrInvalidTransition) - } - if expiresAt.Before(time.Now()) { - return nil, fmt.Errorf("%w: approval token expired", domain.ErrInvalidTransition) - } - if *tokenHash != hashToken(*req.Body.Token) { - return nil, fmt.Errorf("%w: invalid approval token", domain.ErrInvalidInput) - } + var token string + if req.Body.Token != nil { + token = *req.Body.Token } - // Map decision to status. - var status string - switch req.Body.Decision { - case gen.Approve: - status = "approved" - case gen.Deny: - status = "denied" - case gen.Revoke: - status = "revoked" - default: - return nil, fmt.Errorf("%w: invalid decision %q", domain.ErrInvalidInput, req.Body.Decision) - } - - if err := q.UpdateApprovalStatus(ctx, sqlcgen.UpdateApprovalStatusParams{ - EntityID: id, - Status: status, - }); err != nil { - if err == pgx.ErrNoRows { - return nil, fmt.Errorf("%w: approval %s not found or already decided", domain.ErrNotFound, req.Id) - } - return nil, err - } - - // Re-read approval. - app, err := q.GetApprovalByID(ctx, id) + result, err := s.approvalSvc.Decide(ctx, app.ApprovalDecideCmd{ + ApprovalID: domain.UUID(id.String()), + Token: token, + Decision: string(req.Body.Decision), + Actor: actor, + }) if err != nil { return nil, err } - approval := approvalToGen(app) - - if auditErr := observability.Audit(ctx, q, actorType, actor, "decide", - &id, "POST", "/api/v1/approvals/"+req.Id+"/decision", "", - nil, - map[string]any{"decision": status}); auditErr != nil { - return nil, auditErr + // Approve: dispatch the SSH work the repo just un-gated, in the + // background so the HTTP response is not blocked by the execution. + if result.Resume != nil { + res := result.Resume + safego.Go("httpapi:executeApprovedAction", func() { + s.executeApprovedAction(context.Background(), s.pool, uuid.MustParse(string(res.ExecutionID)), res.TargetSlug, res.Action) + }) + slog.Info("httpapi: approved execution queued", + "execution_id", res.ExecutionID, "target", res.TargetSlug, "action", res.Action) } - // Emit for SSE fan-out (in-tx; NOTIFY fires post-commit). - if evErr := observability.Event(ctx, q, "approval.decided", &id, "info", "api", "", - map[string]any{"decision": status, "actor": actor}); evErr != nil { - return nil, evErr - } - - // On approve: execute the linked gated command. - if status == "approved" { - var execID, targetID uuid.UUID - var actionStr, targetSlug, riskClass string - err := tx.QueryRow(ctx, ` - SELECT e.entity_id, e.target_entity_id, e.action, e.risk_class - FROM executions e - WHERE e.approval_id = $1 AND e.status = 'pending_approval' - LIMIT 1`, id).Scan(&execID, &targetID, &actionStr, &riskClass) - if err == nil { - // Resolve target entity slug from targetID. - _ = tx.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", targetID).Scan(&targetSlug) - - safego.Go("httpapi:executeApprovedAction", func() { - s.executeApprovedAction(context.Background(), s.pool, execID, targetSlug, actionStr) - }) - // Status only — risk_class was set correctly at request time - // (e.g. by policy.ClassifyCommand for `run`); overwriting it to - // a hardcoded 'config_mutation' here corrupted the audit ledger - // for every other risk class, including destructive. - _, _ = tx.Exec(ctx, `UPDATE executions SET status = 'approved' WHERE entity_id = $1`, execID) - - // Approving a plan step — by ANY route (this endpoint backs both - // the chat Approve button and chat-assent) — opens/extends the - // agent's assent window. This is the scope gate the Nomos - // auto-continuation worker checks: with the window open, the - // finished execution's result is fed back to the agent so it runs - // the plan to completion. Without opening it here, approving via - // the button (instead of typing "go ahead") would silently not - // auto-continue. - var agentID *uuid.UUID - if qerr := tx.QueryRow(ctx, "SELECT agent_id FROM executions WHERE entity_id = $1", execID).Scan(&agentID); qerr == nil && agentID != nil { - expires := time.Now().Add(30 * time.Minute).UTC().Format(time.RFC3339) - _, _ = tx.Exec(ctx, `INSERT INTO autonomy_settings (key, value) VALUES ($1, $2) - ON CONFLICT (key) DO UPDATE SET value = $2`, "assent_window.agent:"+agentID.String(), expires) - - // Approving a DESTRUCTIVE step via the button is exactly as - // explicit as a typed "I confirm" — the operator affirmatively - // clicked Approve on a card that said DESTRUCTIVE. Open the - // same short, target-scoped destructive window chat-assent's - // typed-confirm path opens, for parity: a multi-step - // destructive recovery (stop, then destroy) shouldn't need a - // fresh confirmation per click any more than it needs one per - // typed phrase. - if riskClass == "destructive" && targetSlug != "" { - dExpires := time.Now().Add(15 * time.Minute).UTC().Format(time.RFC3339) - _, _ = tx.Exec(ctx, `INSERT INTO autonomy_settings (key, value) VALUES ($1, $2) - ON CONFLICT (key) DO UPDATE SET value = $2`, - "destructive_window.agent:"+agentID.String()+".target:"+targetSlug, dExpires) - } - } - - slog.Info("httpapi: approved execution queued", - "execution_id", execID, "target", targetSlug, "action", actionStr) - } else { - slog.Warn("httpapi: no pending execution found for approval", "approval_id", id, "error", err) - } - } else { - // Denied/revoked: reflect it on the linked execution too. Previously - // only the approvals row changed, so the execution stayed - // 'pending_approval' forever — any UI/poller reading execution - // status (not approval status) never saw the decision. - _, _ = tx.Exec(ctx, `UPDATE executions SET status = $2, completed_at = now() WHERE approval_id = $1 AND status = 'pending_approval'`, id, status) - } - - // If this execution belongs to a nomos session, flip it out of - // awaiting_input — the counterpart to classifyAndGate flipping it IN - // the moment the approval was created (internal/mcp/server.go's - // markSessionAwaitingApproval). Runs for all three decisions (approve/ - // deny/revoke): each one is an operator answer to "what do I do about - // this?", same as answerQuestion's unconditional resume-to-executing - // (cmd/nomos/store.go) for a session_questions answer. - var awaitingSessionID string - _ = tx.QueryRow(ctx, ` - SELECT pe.session_id FROM nomos_plan_executions pe - JOIN executions ex ON ex.entity_id = pe.execution_id - WHERE ex.approval_id = $1 - LIMIT 1`, id).Scan(&awaitingSessionID) - if awaitingSessionID != "" { - if rtag, rerr := tx.Exec(ctx, ` - UPDATE agent_sessions SET status = 'executing', last_active_at = now() - WHERE id = $1 AND status = 'awaiting_input'`, awaitingSessionID); rerr == nil && rtag.RowsAffected() > 0 { - var taskEntID *uuid.UUID - var e uuid.UUID - if qerr := tx.QueryRow(ctx, `SELECT entity_id FROM agent_sessions WHERE id = $1`, awaitingSessionID).Scan(&e); qerr == nil && e != uuid.Nil { - taskEntID = &e - } - _ = observability.Event(ctx, q, "task.status", taskEntID, "info", "api", awaitingSessionID, - map[string]any{"status": "executing", "reason": "approval_decided", "decision": status}) - } - } - - if err := tx.Commit(ctx); err != nil { - return nil, err - } - - return gen.DecideApproval200JSONResponse(approval), nil + return gen.DecideApproval200JSONResponse(approvalToGenApproval(result.Approval)), nil } -func approvalToGen(a sqlcgen.Approval) gen.Approval { +// approvalToGenApproval maps the domain approval onto the wire shape. The +// subject slug is not carried on the domain type, so it is left empty on the +// decide path. +func approvalToGenApproval(a domain.Approval) gen.Approval { app := gen.Approval{ - Id: a.EntityID, + Id: uuid.MustParse(string(a.EntityID)), Action: a.Action, RiskClass: a.RiskClass, Kind: gen.ApprovalKind(a.Kind), @@ -275,12 +135,13 @@ func approvalToGen(a sqlcgen.Approval) gen.Approval { DecidedAt: a.DecidedAt, CreatedAt: a.CreatedAt, } - if a.DecidedBy != nil { - s := a.DecidedBy.String() - app.DecidedBy = &s + if a.DecidedBy != "" { + u := uuid.MustParse(string(a.DecidedBy)) + str := u.String() + app.DecidedBy = &str } - var payload map[string]any - if len(a.Payload) > 0 && json.Unmarshal(a.Payload, &payload) == nil && len(payload) > 0 { + if len(a.Payload) > 0 { + payload := map[string]any(a.Payload) app.Payload = &payload } return app diff --git a/internal/httpapi/execution_logs.go b/internal/httpapi/execution_logs.go index 61dcdf2b..f5175724 100644 --- a/internal/httpapi/execution_logs.go +++ b/internal/httpapi/execution_logs.go @@ -6,7 +6,7 @@ import ( "strconv" "strings" - "github.com/dtoro/oikos/internal/execlog" + "github.com/dtoro/oikos/internal/adapters/postgres" "github.com/go-chi/chi/v5" "github.com/google/uuid" ) @@ -34,7 +34,7 @@ func (s *Server) serveExecutionLogs(w http.ResponseWriter, req *http.Request) { } } - chunks, err := execlog.Read(ctx, s.pool, execID, limit) + chunks, err := db.ReadExecutionLog(ctx, s.pool, execID, limit) if err != nil { writeProblem(w, req, http.StatusInternalServerError, "query failed", err.Error()) return diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 49c4ccec..77050cc5 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -24,8 +24,8 @@ import ( "time" "github.com/dtoro/oikos/internal/actuator" - "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/adapters/postgres" + "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/core/app" "github.com/dtoro/oikos/internal/core/ports" "github.com/dtoro/oikos/internal/httpapi/gen" @@ -65,10 +65,10 @@ type Server struct { // entities is the entity-aggregate use-case service (Phase 3 of the // hexagonal refactor); entityRepo exposes the idempotency reads the // replay path needs. Wired here until main becomes the composition root. - entities *app.EntityService - entityRepo *db.EntityRepo - readModels ports.ReadModels - relService *app.RelationshipService + entities *app.EntityService + entityRepo *db.EntityRepo + readModels ports.ReadModels + relService *app.RelationshipService // provisioning owns the pct_create flow (Phase 7): spec defaults, // provisioner dispatch, guest registration in the graph. provisioning *app.ProvisioningService @@ -77,6 +77,10 @@ type Server struct { // execSvc is the run-submission use-case (Phase 4): PolicyService // gating + auto-run/queue dispatch, shared by the MCP tools. execSvc *app.ExecutionService + // approvalSvc is the approval-decision use-case (Phase 9): token + // verification + execution un-gating, shared by the REST decision + // endpoint and the MCP decide_approval path. + approvalSvc *app.ApprovalService } // NewHandler builds the full HTTP handler: /healthz (unauthenticated, @@ -86,7 +90,7 @@ type Server struct { // holds a dedicated pooled connection for LISTEN. Callers MUST cancel ctx // before closing the pool — otherwise the held connection never releases // and pool.Close() deadlocks. -func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config, entities *app.EntityService, entityRepo *db.EntityRepo, readModels ports.ReadModels, relService *app.RelationshipService, provisioning *app.ProvisioningService, seeds *app.SeedService, execSvc *app.ExecutionService) http.Handler { +func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config, entities *app.EntityService, entityRepo *db.EntityRepo, readModels ports.ReadModels, relService *app.RelationshipService, provisioning *app.ProvisioningService, seeds *app.SeedService, execSvc *app.ExecutionService, approvalSvc *app.ApprovalService) http.Handler { s := &Server{ pool: pool, cfg: cfg, @@ -100,6 +104,7 @@ func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config, entities provisioning: provisioning, seeds: seeds, execSvc: execSvc, + approvalSvc: approvalSvc, } // Wire secrets backend: Infisical primary with SOPS DR fallback. @@ -948,10 +953,10 @@ main(); // ListenAndServe runs the API server with graceful shutdown on ctx cancel // (SG4): stop accepting, drain in-flight for up to 30s, then exit. -func ListenAndServe(ctx context.Context, pool *db.Pool, cfg config.Config, entities *app.EntityService, entityRepo *db.EntityRepo, readModels ports.ReadModels, relService *app.RelationshipService, provisioning *app.ProvisioningService, seeds *app.SeedService, execSvc *app.ExecutionService) error { +func ListenAndServe(ctx context.Context, pool *db.Pool, cfg config.Config, entities *app.EntityService, entityRepo *db.EntityRepo, readModels ports.ReadModels, relService *app.RelationshipService, provisioning *app.ProvisioningService, seeds *app.SeedService, execSvc *app.ExecutionService, approvalSvc *app.ApprovalService) error { srv := &http.Server{ Addr: cfg.APIListen, - Handler: NewHandler(ctx, pool, cfg, entities, entityRepo, readModels, relService, provisioning, seeds, execSvc), + Handler: NewHandler(ctx, pool, cfg, entities, entityRepo, readModels, relService, provisioning, seeds, execSvc, approvalSvc), ReadHeaderTimeout: 10 * time.Second, } diff --git a/internal/mcp/discover.go b/internal/mcp/discover.go index d85385f0..84577ef4 100644 --- a/internal/mcp/discover.go +++ b/internal/mcp/discover.go @@ -6,7 +6,6 @@ import ( "strings" "github.com/dtoro/oikos/internal/adapters/postgres" - "github.com/dtoro/oikos/internal/execlog" "github.com/dtoro/oikos/internal/remote" ) @@ -34,10 +33,10 @@ func discoverInfraDrift(ctx context.Context, pool *db.Pool) any { // 2. DB guests keyed by pve_id. type dbGuest struct { - Slug string `json:"slug"` - Type string `json:"type"` - PveID string `json:"pve_id"` - Host string `json:"host"` + Slug string `json:"slug"` + Type string `json:"type"` + PveID string `json:"pve_id"` + Host string `json:"host"` } dbGuests := map[string]dbGuest{} gr, err := pool.Query(ctx, @@ -65,7 +64,7 @@ func discoverInfraDrift(ctx context.Context, pool *db.Pool) any { } for _, cmd := range []string{"pct list", "qm list"} { out, eerr := sshExecStream(ctx, et.Host, et.User, et.Wrap(cmd), - execlog.Sink(func(string, []byte) {})) + db.ExecLogSink(func(string, []byte) {})) if eerr != nil { hostErrors[hs+" "+cmd] = eerr.Error() continue @@ -88,12 +87,12 @@ func discoverInfraDrift(ctx context.Context, pool *db.Pool) any { } return map[string]any{ - "hosts_queried": len(hosts), - "live_guests": len(live), - "db_guests": len(dbGuests), + "hosts_queried": len(hosts), + "live_guests": len(live), + "db_guests": len(dbGuests), "missing_entities": missing, // in Proxmox, no DB entity - "ghost_entities": ghost, // in DB, not live in Proxmox - "host_errors": hostErrors, + "ghost_entities": ghost, // in DB, not live in Proxmox + "host_errors": hostErrors, } } diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 3ef0af76..91309010 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -20,12 +20,11 @@ import ( "time" "github.com/dtoro/oikos/internal/actuator" + "github.com/dtoro/oikos/internal/adapters/postgres" + "github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen" "github.com/dtoro/oikos/internal/core/app" "github.com/dtoro/oikos/internal/core/domain" "github.com/dtoro/oikos/internal/core/ports" - "github.com/dtoro/oikos/internal/adapters/postgres" - "github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen" - "github.com/dtoro/oikos/internal/execlog" "github.com/dtoro/oikos/internal/observability" "github.com/dtoro/oikos/internal/policy" "github.com/dtoro/oikos/internal/remote" @@ -471,7 +470,7 @@ func sshExec(ctx context.Context, host, user, command string) (string, error) { // sshExecStream runs a command and reports its combined output, forwarding // each chunk to sink as it arrives. A nil sink behaves exactly as before. -func sshExecStream(ctx context.Context, host, user, command string, sink execlog.Sink) (string, error) { +func sshExecStream(ctx context.Context, host, user, command string, sink db.ExecLogSink) (string, error) { initSSH() if len(sshKey) == 0 { return "", fmt.Errorf("no SSH key available") @@ -619,7 +618,7 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, execSvc *app.ExecutionS SessionID: sessionID, Async: app.IsLongRunningCommand(command), SinkFactory: func(ctx context.Context, execID domain.UUID, correlationID string) (func(string, []byte), func()) { - return execlog.New(ctx, pool, uuid.MustParse(string(execID)), correlationID) + return db.NewExecutionLog(ctx, pool, uuid.MustParse(string(execID)), correlationID) }, }) return renderSubmit(targetSlug, command, res) diff --git a/internal/nomos/assent/assent_test.go b/internal/nomos/assent/assent_test.go index 1bc41140..a69f7778 100644 --- a/internal/nomos/assent/assent_test.go +++ b/internal/nomos/assent/assent_test.go @@ -11,8 +11,8 @@ func TestIsAssent_Positive(t *testing.T) { "ok go ahead and run it", } for _, c := range cases { - if !isAssent(c) { - t.Errorf("isAssent(%q) = false, want true", c) + if !IsAssent(c) { + t.Errorf("IsAssent(%q) = false, want true", c) } } } @@ -24,8 +24,8 @@ func TestIsAssent_Negative(t *testing.T) { "maybe later", "", } for _, c := range cases { - if isAssent(c) { - t.Errorf("isAssent(%q) = true, want false", c) + if IsAssent(c) { + t.Errorf("IsAssent(%q) = true, want false", c) } } } @@ -38,8 +38,8 @@ func TestIsAssent_NegationBeatsAssentWord(t *testing.T) { "wait, not yet please", } for _, c := range cases { - if isAssent(c) { - t.Errorf("isAssent(%q) = true, want false (negation should block)", c) + if IsAssent(c) { + t.Errorf("IsAssent(%q) = true, want false (negation should block)", c) } } } @@ -47,7 +47,7 @@ func TestIsAssent_NegationBeatsAssentWord(t *testing.T) { // TestIsAssent_WholeWordBoundary regression-tests a real false positive found // live: the old substring check matched "yes" inside "yesterday" (and would // equally match "confirm" inside "confirmed"/"unconfirmed" for -// isTypedConfirmation below) because only negation used a word-boundary +// IsTypedConfirmation below) because only negation used a word-boundary // check — assent/confirm words used a bare strings.Contains. Confirmed via a // throwaway probe before being fixed; kept here permanently so a future // change can't silently reintroduce it. @@ -57,14 +57,14 @@ func TestIsAssent_WholeWordBoundary(t *testing.T) { "my eyesight isn't great, what does that say", } for _, c := range cases { - if isAssent(c) { - t.Errorf("isAssent(%q) = true, want false (word-boundary: 'yes' must not match inside 'yesterday'/'eyesight')", c) + if IsAssent(c) { + t.Errorf("IsAssent(%q) = true, want false (word-boundary: 'yes' must not match inside 'yesterday'/'eyesight')", c) } } } // TestIsTypedConfirmation_ContractedNegation regression-tests the other real -// false positive: isTypedConfirmation gates DESTRUCTIVE actions, and +// false positive: IsTypedConfirmation gates DESTRUCTIVE actions, and // "confirm" matching inside "confirmed" combined with contracted negatives // ("haven't") not being in negationWords meant a message that explicitly // says the operator has NOT confirmed something could read as confirming it. @@ -75,8 +75,8 @@ func TestIsTypedConfirmation_ContractedNegation(t *testing.T) { "we can't confirm that until tomorrow", } for _, c := range cases { - if isTypedConfirmation(c) { - t.Errorf("isTypedConfirmation(%q) = true, want false (contracted negation should block)", c) + if IsTypedConfirmation(c) { + t.Errorf("IsTypedConfirmation(%q) = true, want false (contracted negation should block)", c) } } } @@ -89,8 +89,8 @@ func TestIsTypedConfirmation(t *testing.T) { "yes I confirm", } for _, c := range positive { - if !isTypedConfirmation(c) { - t.Errorf("isTypedConfirmation(%q) = false, want true", c) + if !IsTypedConfirmation(c) { + t.Errorf("IsTypedConfirmation(%q) = false, want true", c) } } negative := []string{ @@ -98,8 +98,8 @@ func TestIsTypedConfirmation(t *testing.T) { "no, don't confirm yet", "wait", "", } for _, c := range negative { - if isTypedConfirmation(c) { - t.Errorf("isTypedConfirmation(%q) = true, want false (only explicit confirm should pass)", c) + if IsTypedConfirmation(c) { + t.Errorf("IsTypedConfirmation(%q) = true, want false (only explicit confirm should pass)", c) } } } diff --git a/internal/nomos/session/store.go b/internal/nomos/session/store.go index ae47ee22..928e36e7 100644 --- a/internal/nomos/session/store.go +++ b/internal/nomos/session/store.go @@ -1,4 +1,4 @@ -package main +package session import ( "context" @@ -14,30 +14,31 @@ import ( "github.com/dtoro/oikos/internal/observability" "github.com/google/uuid" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" ) const maxToolResultSize = 4096 -// errPlanInFlight is returned by proposePlan when called again after a step +// ErrPlanInFlight is returned by proposePlan when called again after a step // has already started. The agent must advance the existing plan with // update_plan_step + run instead of re-proposing — re-proposing was the // source of duplicate plans in the sidebar (operator-reported 2026-07-14). // The caller translates this into a directive tool result. -var errPlanInFlight = errors.New("plan already in flight") +var ErrPlanInFlight = errors.New("plan already in flight") -// errPlanStepNotFound is returned by updatePlanStep when no step matches the +// ErrPlanStepNotFound is returned by updatePlanStep when no step matches the // given seq in the CURRENT (MAX) generation — either the seq is out of range, // or (after a re-plan) the model addressed a stale 1-based number. seq is // generation-relative, so this never resurrects a superseded generation's row. // The caller translates it into a directive tool result (P0.1). -var errPlanStepNotFound = errors.New("plan step not found in current generation") +var ErrPlanStepNotFound = errors.New("plan step not found in current generation") -type store struct { +type Store struct { pool *pgxpool.Pool } -func newStore(ctx context.Context, databaseURL string) (*store, error) { +func New(ctx context.Context, databaseURL string) (*Store, error) { if databaseURL == "" { return nil, nil } @@ -49,8 +50,8 @@ func newStore(ctx context.Context, databaseURL string) (*store, error) { pool.Close() return nil, fmt.Errorf("ping db: %w", err) } - s := &store{pool: pool} - s.cleanupStaleExecutions(ctx, time.Hour) + s := &Store{pool: pool} + s.CleanupStaleExecutions(ctx, time.Hour) return s, nil } @@ -62,7 +63,7 @@ func newStore(ctx context.Context, databaseURL string) (*store, error) { // `running` and `pending_approval` executions pile up in the DB and pollute // the Operations page + session rail badges. Called at startup (maxAge=1h) // and periodically (maxAge=10m) by the sweep worker. -func (s *store) cleanupStaleExecutions(ctx context.Context, maxAge time.Duration) int { +func (s *Store) CleanupStaleExecutions(ctx context.Context, maxAge time.Duration) int { if s == nil { return 0 } @@ -85,12 +86,21 @@ func (s *store) cleanupStaleExecutions(ctx context.Context, maxAge time.Duration return n } -func (s *store) close() { +func (s *Store) Close() { if s.pool != nil { s.pool.Close() } } +// Exec runs a raw SQL query against the store's pool. Used by the agent to +// write autonomy_settings rows directly. +func (s *Store) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) { + if s == nil { + return pgconn.CommandTag{}, nil + } + return s.pool.Exec(ctx, sql, args...) +} + // session is a chat session elevated to a task: goal-structured work with a // lifecycle status and an outcome (see migration 018 / the task-board plan). // Outcome/Summary/EntityID are empty until set, hence omitempty. @@ -104,7 +114,7 @@ func (s *store) close() { // Blocker is a short structured reason: approval_timeout, // classifier_overreach, user_abandoned, tool_error, etc. See // plans/2026-07-20-session-review-ten-sessions.md P1.5. -type session struct { +type Session struct { ID string `json:"id"` Title string `json:"title"` Actor string `json:"actor"` @@ -129,7 +139,7 @@ type session struct { DurationSeconds int `json:"duration_seconds,omitempty"` } -type message struct { +type Message struct { ID string `json:"id"` SessionID string `json:"session_id"` Role string `json:"role"` @@ -137,9 +147,9 @@ type message struct { CreatedAt time.Time `json:"created_at"` } -func (s *store) createSession(ctx context.Context, title string) (*session, error) { +func (s *Store) CreateSession(ctx context.Context, title string) (*Session, error) { if s == nil { - return &session{ID: "ephemeral", Title: title, Actor: "agent:nomos", Status: "active"}, nil + return &Session{ID: "ephemeral", Title: title, Actor: "agent:nomos", Status: "active"}, nil } var id string err := s.pool.QueryRow(ctx, @@ -152,7 +162,7 @@ func (s *store) createSession(ctx context.Context, title string) (*session, erro // off the existing relationships graph. Best-effort: a failure here must not // block the chat — the session is usable without a graph anchor. entityID := s.createTaskEntity(ctx, id, title) - return &session{ID: id, Title: title, Actor: "agent:nomos", Status: "active", + return &Session{ID: id, Title: title, Actor: "agent:nomos", Status: "active", EntityID: entityID, CreatedAt: time.Now(), LastActiveAt: time.Now()}, nil } @@ -160,7 +170,7 @@ func (s *store) createSession(ctx context.Context, title string) (*session, erro // anchors this task's knowledge and involved-entity relationships, and records // it on the session. Returns the entity id, or "" on failure — non-fatal, see // caller. Requires the 'task' entity type (seeds/ontology.yaml). -func (s *store) createTaskEntity(ctx context.Context, sessionID, title string) string { +func (s *Store) createTaskEntity(ctx context.Context, sessionID, title string) string { entityID, _ := uuid.NewV7() slug := "task:" + sessionID // name is UNIQUE(type,name) and chat titles collide ("hi" ×6), so key the @@ -191,7 +201,7 @@ func (s *store) createTaskEntity(ctx context.Context, sessionID, title string) s return entityID.String() } -func (s *store) saveMessage(ctx context.Context, sessionID, role string, content json.RawMessage) error { +func (s *Store) SaveMessage(ctx context.Context, sessionID, role string, content json.RawMessage) error { if s == nil { return nil } @@ -208,7 +218,7 @@ func (s *store) saveMessage(ctx context.Context, sessionID, role string, content // the worker inserts a placeholder immediately and updates the SAME row as // each tool call completes, so a poller sees individual steps land, not just // a final rolled-up summary. -func (s *store) insertMessageReturningID(ctx context.Context, sessionID, role string, content json.RawMessage) (uuid.UUID, error) { +func (s *Store) InsertMessageReturningID(ctx context.Context, sessionID, role string, content json.RawMessage) (uuid.UUID, error) { if s == nil { return uuid.Nil, nil } @@ -219,7 +229,7 @@ func (s *store) insertMessageReturningID(ctx context.Context, sessionID, role st return id, err } -func (s *store) updateMessage(ctx context.Context, id uuid.UUID, content json.RawMessage) error { +func (s *Store) UpdateMessage(ctx context.Context, id uuid.UUID, content json.RawMessage) error { if s == nil || id == uuid.Nil { return nil } @@ -234,19 +244,19 @@ func (s *store) updateMessage(ctx context.Context, id uuid.UUID, content json.Ra // retries failed), the placeholder row is deleted instead of persisting an // empty assistant bubble — the error was already streamed to the frontend // via the 'done with error=true' event, so the operator sees it inline. -func (s *store) deleteMessage(ctx context.Context, id uuid.UUID) { +func (s *Store) DeleteMessage(ctx context.Context, id uuid.UUID) { if s == nil || id == uuid.Nil { return } s.pool.Exec(ctx, `DELETE FROM agent_messages WHERE id = $1`, id) } -// lastUserMessage returns the most recent user message text for a session, +// lastUserMessage returns the most recent user message text for a Session, // or "" if none. Used to build a context-rich reconnect/resume note: instead // of a generic "report your state," the note can say "the operator's last // message was X — advance the plan" so the agent doesn't re-propose or // re-execute on a reconnect (the operator-reported 2026-07-14 divergence). -func (s *store) lastUserMessage(ctx context.Context, sessionID string) string { +func (s *Store) lastUserMessage(ctx context.Context, sessionID string) string { if s == nil || sessionID == "" || sessionID == "ephemeral" { return "" } @@ -272,7 +282,7 @@ func (s *store) lastUserMessage(ctx context.Context, sessionID string) string { // the plan with update_plan_step + run" instead of the generic "report // your state" (which caused the agent to re-propose and duplicate the plan // in the sidebar — operator-reported 2026-07-14). -func (s *store) hasPlanInFlight(ctx context.Context, sessionID string) bool { +func (s *Store) HasPlanInFlight(ctx context.Context, sessionID string) bool { if s == nil || sessionID == "" || sessionID == "ephemeral" { return false } @@ -292,12 +302,12 @@ func (s *store) hasPlanInFlight(ctx context.Context, sessionID string) bool { // duplicate the plan on a reconnect (operator-reported 2026-07-14); this // enrichment gives the agent enough context to do the right thing even // through the reconnect path. -func (s *store) enrichResumeNote(ctx context.Context, sessionID, base string) string { +func (s *Store) EnrichResumeNote(ctx context.Context, sessionID, base string) string { if s == nil || sessionID == "" || sessionID == "ephemeral" { return base } last := s.lastUserMessage(ctx, sessionID) - inFlight := s.hasPlanInFlight(ctx, sessionID) + inFlight := s.HasPlanInFlight(ctx, sessionID) if last == "" && !inFlight { return base } @@ -346,21 +356,21 @@ func truncateToolResults(content json.RawMessage) json.RawMessage { return out } -func (s *store) touchSession(ctx context.Context, id string) { +func (s *Store) TouchSession(ctx context.Context, id string) { if s != nil { s.pool.Exec(ctx, `UPDATE agent_sessions SET last_active_at=now() WHERE id=$1`, id) } } -func (s *store) listSessions(ctx context.Context) ([]session, error) { - return s.listSessionsFiltered(ctx, listFilter{Limit: 50}) +func (s *Store) ListSessions(ctx context.Context) ([]Session, error) { + return s.ListSessionsFiltered(ctx, ListFilter{Limit: 50}) } // listFilter carries the optional WHERE/ORDER clauses added by P2.8 // (filtering & pagination). All fields optional; empty values are no-ops. // The handler in main.go parses query params into this struct so the SQL // builder here is the single source of truth for what filters exist. -type listFilter struct { +type ListFilter struct { Outcome string // exact match on outcome (success/partial/failure) Status string // exact match on status (active/done/failed/executing) EntityID string // exact match on entity_id (UUID) @@ -370,7 +380,7 @@ type listFilter struct { Limit int // default 50, clamped by the handler } -func (s *store) listSessionsFiltered(ctx context.Context, f listFilter) ([]session, error) { +func (s *Store) ListSessionsFiltered(ctx context.Context, f ListFilter) ([]Session, error) { if s == nil { return nil, nil } @@ -474,9 +484,9 @@ func (s *store) listSessionsFiltered(ctx context.Context, f listFilter) ([]sessi } defer rows.Close() - var out []session + var out []Session for rows.Next() { - var sess session + var sess Session if err := rows.Scan(&sess.ID, &sess.Title, &sess.Actor, &sess.Goal, &sess.Status, &sess.Outcome, &sess.Summary, &sess.EntityID, &sess.PendingApprovals, &sess.Blocker, &sess.CreatedAt, &sess.LastActiveAt, &sess.ClosedAt, @@ -488,11 +498,11 @@ func (s *store) listSessionsFiltered(ctx context.Context, f listFilter) ([]sessi return out, rows.Err() } -func (s *store) getSession(ctx context.Context, id string) (*session, error) { +func (s *Store) GetSession(ctx context.Context, id string) (*Session, error) { if s == nil { return nil, nil } - var sess session + var sess Session err := s.pool.QueryRow(ctx, `SELECT s.id, s.title, s.actor, s.goal, s.status, COALESCE(s.outcome, ''), s.summary, COALESCE(s.entity_id::text, ''), 0, COALESCE(s.blocker, ''), @@ -533,7 +543,7 @@ func (s *store) getSession(ctx context.Context, id string) (*session, error) { // prior session's goal + summary at set_goal time lets the agent pick up the // thread instead of rediscovering it. See // plans/2026-07-20-session-review-ten-sessions.md P1.3. -func (s *store) recentPartialSessions(ctx context.Context, excludeSessionID string, since time.Duration) ([]session, error) { +func (s *Store) RecentPartialSessions(ctx context.Context, excludeSessionID string, since time.Duration) ([]Session, error) { if s == nil { return nil, nil } @@ -576,9 +586,9 @@ func (s *store) recentPartialSessions(ctx context.Context, excludeSessionID stri } defer rows.Close() - var out []session + var out []Session for rows.Next() { - var sess session + var sess Session if err := rows.Scan(&sess.ID, &sess.Title, &sess.Actor, &sess.Goal, &sess.Status, &sess.Outcome, &sess.Summary, &sess.EntityID, &sess.PendingApprovals, &sess.Blocker, &sess.CreatedAt, &sess.LastActiveAt, &sess.ClosedAt, @@ -596,7 +606,7 @@ func (s *store) recentPartialSessions(ctx context.Context, excludeSessionID stri // it's run. For LLM replay, see getRecentMessages: sending the operator's // full transcript is fine; sending the model's full transcript on every // single turn is not (see getRecentMessages's doc comment). -func (s *store) getMessages(ctx context.Context, sessionID string) ([]message, error) { +func (s *Store) GetMessages(ctx context.Context, sessionID string) ([]Message, error) { if s == nil { return nil, nil } @@ -608,9 +618,9 @@ func (s *store) getMessages(ctx context.Context, sessionID string) ([]message, e } defer rows.Close() - var out []message + var out []Message for rows.Next() { - var m message + var m Message if err := rows.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &m.CreatedAt); err != nil { return nil, err } @@ -644,11 +654,11 @@ type SessionToolCall struct { // it a single SQL + Go walk on the server. Each tool_use/tool_result pair // is emitted as two rows (same id, different Type), preserving the // persisted shape — clients that want the merged shape can group by ID. -func (s *store) getSessionToolCalls(ctx context.Context, sessionID string) ([]SessionToolCall, error) { +func (s *Store) GetSessionToolCalls(ctx context.Context, sessionID string) ([]SessionToolCall, error) { if s == nil { return nil, nil } - msgs, err := s.getMessages(ctx, sessionID) + msgs, err := s.GetMessages(ctx, sessionID) if err != nil { return nil, err } @@ -699,7 +709,7 @@ func (s *store) getSessionToolCalls(ctx context.Context, sessionID string) ([]Se // heavily-autonomous tasks (many auto-continuation cycles) this system is // built to run longest. Fetches limit+1 rows to detect "there's more" // without a separate COUNT query. -func (s *store) getRecentMessages(ctx context.Context, sessionID string, limit int) (msgs []message, truncated bool, err error) { +func (s *Store) GetRecentMessages(ctx context.Context, sessionID string, limit int) (msgs []Message, truncated bool, err error) { if s == nil { return nil, false, nil } @@ -712,9 +722,9 @@ func (s *store) getRecentMessages(ctx context.Context, sessionID string, limit i } defer rows.Close() - var out []message + var out []Message for rows.Next() { - var m message + var m Message if err := rows.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &m.CreatedAt); err != nil { return nil, false, err } @@ -736,7 +746,7 @@ func (s *store) getRecentMessages(ctx context.Context, sessionID string, limit i return out, truncated, nil } -func (s *store) deleteSession(ctx context.Context, id string) error { +func (s *Store) DeleteSession(ctx context.Context, id string) error { if s == nil { return nil } @@ -763,9 +773,9 @@ func (s *store) deleteSession(ctx context.Context, id string) error { return nil } -// taskEntityPtr returns the task entity id for a session, or nil — used as the +// taskEntityPtr returns the task entity id for a Session, or nil — used as the // entity_id on task-scoped events so they anchor to the task in the graph. -func (s *store) taskEntityPtr(ctx context.Context, sessionID string) *uuid.UUID { +func (s *Store) taskEntityPtr(ctx context.Context, sessionID string) *uuid.UUID { var id uuid.UUID if err := s.pool.QueryRow(ctx, `SELECT entity_id FROM agent_sessions WHERE id = $1`, sessionID).Scan(&id); err != nil || id == uuid.Nil { @@ -795,7 +805,7 @@ func (s *store) taskEntityPtr(ctx context.Context, sessionID string) *uuid.UUID // sessions.md P1.4 (session 55927f0a had two set_goal calls with the first // implicitly abandoned when the operator said "lets just keep ludo-library // then"). -func (s *store) setGoal(ctx context.Context, sessionID, goal string) error { +func (s *Store) SetGoal(ctx context.Context, sessionID, goal string) error { if s == nil || sessionID == "" || sessionID == "ephemeral" { return nil } @@ -845,7 +855,7 @@ func (s *store) setGoal(ctx context.Context, sessionID, goal string) error { // // Returns true if the session was actually reopened (was terminal), false if // it was already active (no-op). -func (s *store) reopenSession(ctx context.Context, sessionID string) bool { +func (s *Store) ReopenSession(ctx context.Context, sessionID string) bool { if s == nil || sessionID == "" || sessionID == "ephemeral" { return false } @@ -874,7 +884,7 @@ func (s *store) reopenSession(ctx context.Context, sessionID string) bool { } // planStepInput is one step as the agent proposes it. -type planStepInput struct { +type PlanStepInput struct { Title string Detail string TargetSlug string @@ -895,7 +905,7 @@ type planStepInput struct { // Refusing is the correct default — the tool result tells the agent how // to advance, and the generation column tracks revisions if a genuine // re-plan is ever allowed. -func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planStepInput) ([]map[string]any, error) { +func (s *Store) ProposePlan(ctx context.Context, sessionID string, steps []PlanStepInput) ([]map[string]any, error) { if s == nil || sessionID == "" || sessionID == "ephemeral" { return nil, nil } @@ -920,7 +930,7 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS // A plan is already in flight (a step is running/done/failed/...). // Refuse the re-proposal — the agent must advance with // update_plan_step + run. The caller surfaces a directive. - return nil, errPlanInFlight + return nil, ErrPlanInFlight } // Fresh/revise: mark any prior PENDING steps as `replaced` (not DELETE). // The rows are kept for the generation counter (MAX(generation)+1 below) @@ -992,7 +1002,7 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS // be marked complete while an earlier step is still pending, preventing the // agent from marking step 5 done before step 4 (observed in production: the // agent rushed to close all steps in a final turn, in reverse order). -func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, status, execID, replacedReason string) error { +func (s *Store) UpdatePlanStep(ctx context.Context, sessionID string, seq int, status, execID, replacedReason string) error { if s == nil || sessionID == "" || sessionID == "ephemeral" { return nil } @@ -1011,7 +1021,7 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s return err } if curGen == 0 { - return errPlanStepNotFound + return ErrPlanStepNotFound } stamp := "" switch status { @@ -1044,7 +1054,7 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s // stamp is a fixed literal from the switch above — never user input. // status <> 'replaced' is defense-in-depth: MAX(generation) can't hold a // replaced row, but if it ever could, this refuses the write instead of - // resurrecting it. No matching row → errPlanStepNotFound (stale/out-of-range seq). + // resurrecting it. No matching row → ErrPlanStepNotFound (stale/out-of-range seq). if status == "replaced" && replacedReason != "" { err := s.pool.QueryRow(ctx, ` UPDATE session_plan_steps @@ -1053,7 +1063,7 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s RETURNING id, target_slug`, sessionID, curGen, seq, status, execPtr, replacedReason).Scan(&stepID, &targetSlug) if err != nil { if errors.Is(err, pgx.ErrNoRows) { - return errPlanStepNotFound + return ErrPlanStepNotFound } return err } @@ -1065,7 +1075,7 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s RETURNING id, target_slug`, sessionID, curGen, seq, status, execPtr).Scan(&stepID, &targetSlug) if err != nil { if errors.Is(err, pgx.ErrNoRows) { - return errPlanStepNotFound + return ErrPlanStepNotFound } return err } @@ -1090,21 +1100,21 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s return nil } -// errTaskAlreadyComplete is returned by completeTask when the session is +// ErrTaskAlreadyComplete is returned by completeTask when the session is // already in a terminal state (done/failed/partial). The agent sometimes // re-calls complete_task after a UI clarification (operator-reported // 2026-07-14) — without this guard, the re-completion produces duplicate // knowledge entries and erodes audit-log clarity. The caller translates this // into a directive tool result. -var errTaskAlreadyComplete = errors.New("task already complete") +var ErrTaskAlreadyComplete = errors.New("task already complete") // completeTask sets a task's terminal state, outcome, and one-line summary, // mirrors the outcome onto the task entity's attributes (so the board/graph // show it), and publishes task.status for the live context panel. outcome is // success|failure|partial; status is derived (failure → failed, else done). -// Returns errTaskAlreadyComplete if the session is already terminal — the +// Returns ErrTaskAlreadyComplete if the session is already terminal — the // agent must not re-complete a finished task. -func (s *store) completeTask(ctx context.Context, sessionID, outcome, summary string) error { +func (s *Store) CompleteTask(ctx context.Context, sessionID, outcome, summary string) error { if s == nil || sessionID == "" || sessionID == "ephemeral" { return nil } @@ -1119,7 +1129,7 @@ func (s *store) completeTask(ctx context.Context, sessionID, outcome, summary st // Session doesn't exist or query failed — let the rest of the // function proceed; it'll fail safely on the UPDATE below. } else if currentStatus == "done" || currentStatus == "failed" { - return errTaskAlreadyComplete + return ErrTaskAlreadyComplete } // Auto-cancel any executions still in pending_approval/approved/queued @@ -1271,12 +1281,12 @@ func (s *store) completeTask(ctx context.Context, sessionID, outcome, summary st return nil } -// autoUpsertKnowledge creates a knowledge entry for a completed session, +// autoUpsertKnowledge creates a knowledge entry for a completed Session, // capturing what was done and linking it to the entities involved. Called // automatically from completeTask so every session leaves a trace, even if // the agent forgot to call upsert_knowledge. Only fired for success/partial // outcomes (failures don't have actionable discoveries). -func autoUpsertKnowledge(ctx context.Context, s *store, sessionID, outcome, summary string) { +func autoUpsertKnowledge(ctx context.Context, s *Store, sessionID, outcome, summary string) { var goal string if err := s.pool.QueryRow(ctx, `SELECT COALESCE(goal, '') FROM agent_sessions WHERE id = $1`, @@ -1335,7 +1345,7 @@ func autoUpsertKnowledge(ctx context.Context, s *store, sessionID, outcome, summ // writePlanCompletionRate computes the step completion rate for the current // plan generation and writes it as a task entity attribute so the trend can // be tracked. Baseline from 2026-08-04 audit: 38% (15/39 steps reached done). -func writePlanCompletionRate(ctx context.Context, s *store, sessionID string) { +func writePlanCompletionRate(ctx context.Context, s *Store, sessionID string) { var total, completed int s.pool.QueryRow(ctx, ` SELECT COUNT(*), COALESCE(SUM(CASE WHEN status = 'done' THEN 1 ELSE 0 END), 0) @@ -1359,7 +1369,7 @@ func writePlanCompletionRate(ctx context.Context, s *store, sessionID string) { // last execution, feeding the pattern-extraction pipeline that has been empty // since launch. Only created for success/partial outcomes (failures don't // have a specific execution to tie to). -func autoFeedback(ctx context.Context, s *store, sessionID, outcome, summary string) { +func autoFeedback(ctx context.Context, s *Store, sessionID, outcome, summary string) { // Find the last execution linked to this session. var execID uuid.UUID if err := s.pool.QueryRow(ctx, ` @@ -1418,7 +1428,7 @@ var blockerPatterns = []struct { // "uncategorized" when outcome is partial/failed but no signature matched — // better than "" because the audit needs to know this WAS blocked, just for // an unknown reason. Returns "" for success outcomes (caller checks first). -func deriveBlocker(ctx context.Context, s *store, sessionID, summary string) string { +func deriveBlocker(ctx context.Context, s *Store, sessionID, summary string) string { // Pull the last assistant text — that's where the agent's parting // words explain why it didn't finish. var lastText string @@ -1438,7 +1448,7 @@ func deriveBlocker(ctx context.Context, s *store, sessionID, summary string) str // hadEntityWriteback checks whether this session called update_entity_attributes // or create_relationship — used by complete_task to warn the agent when it // forgot to persist entity facts (the #1 cause of knowledge graph drift). -func (s *store) hadEntityWriteback(ctx context.Context, sessionID string) bool { +func (s *Store) HadEntityWriteback(ctx context.Context, sessionID string) bool { if s == nil || sessionID == "" { return true // fail safe: don't warn when we can't check } @@ -1463,7 +1473,7 @@ func (s *store) hadEntityWriteback(ctx context.Context, sessionID string) bool { // only calls get_entity is a degenerate case (SOUL.md: "Don't invent // attributes that don't exist") and must NOT be blocked. Only sessions that // actually executed against a live target get the writeback gate. -func (s *store) hadDiscovery(ctx context.Context, sessionID string) bool { +func (s *Store) HadDiscovery(ctx context.Context, sessionID string) bool { if s == nil || sessionID == "" { return false // fail safe: don't block when we can't check } @@ -1479,7 +1489,7 @@ func (s *store) hadDiscovery(ctx context.Context, sessionID string) bool { // sessionGoal returns the session's goal text, empty string if not found. // Used by complete_task to check whether the goal involved a reachability // verification before marking success. -func (s *store) sessionGoal(ctx context.Context, sessionID string) string { +func (s *Store) SessionGoal(ctx context.Context, sessionID string) string { if s == nil || sessionID == "" { return "" } @@ -1494,7 +1504,7 @@ func (s *store) sessionGoal(ctx context.Context, sessionID string) string { // reachability in recent turns — ping_service, or a run with curl/wget that // returned successfully. Used by complete_task as a soft warning when the // goal involved a reachability check but no recent verification occurred. -func (s *store) hadRecentVerification(ctx context.Context, sessionID string) bool { +func (s *Store) HadRecentVerification(ctx context.Context, sessionID string) bool { if s == nil || sessionID == "" { return true // fail safe: don't warn when we can't check } @@ -1537,7 +1547,7 @@ type staleGoalSession struct { // alone) but have sat non-terminal past idleThreshold. completion_nudges // tells the caller whether to nudge (0) or give up and auto-close (>=1) — // see processIdleSweep in continue.go. -func (s *store) staleGoalSessions(ctx context.Context, idleThreshold time.Duration, limit int) []staleGoalSession { +func (s *Store) StaleGoalSessions(ctx context.Context, idleThreshold time.Duration, limit int) []staleGoalSession { if s == nil { return nil } @@ -1563,11 +1573,11 @@ func (s *store) staleGoalSessions(ctx context.Context, idleThreshold time.Durati return out } -// bumpCompletionNudge records that the idle sweep nudged a stalled session, +// bumpCompletionNudge records that the idle sweep nudged a stalled Session, // stamping last_active_at so it isn't picked up again until it's genuinely // idle again (a fresh nudge shouldn't fire every tick while the model is // mid-response to the previous one). -func (s *store) bumpCompletionNudge(ctx context.Context, sessionID string) error { +func (s *Store) BumpCompletionNudge(ctx context.Context, sessionID string) error { if s == nil { return nil } @@ -1582,7 +1592,7 @@ func (s *store) bumpCompletionNudge(ctx context.Context, sessionID string) error // still pending or running. Used by autoCompleteIfPlanDone to auto-close a // task when the agent did all the work but forgot to call complete_task. // Returns false if there are no plan steps at all (no plan was proposed). -func (s *store) allPlanStepsTerminal(ctx context.Context, sessionID string) bool { +func (s *Store) AllPlanStepsTerminal(ctx context.Context, sessionID string) bool { if s == nil || sessionID == "" || sessionID == "ephemeral" { return false } @@ -1600,7 +1610,7 @@ func (s *store) allPlanStepsTerminal(ctx context.Context, sessionID string) bool // pending_approval state. Used by autoCompleteIfPlanDone to avoid closing a // session that's blocked waiting for operator approval — the agent hit the // P5 gate and can't continue until the operator responds. -func (s *store) hasPendingApprovals(ctx context.Context, sessionID string) bool { +func (s *Store) HasPendingApprovals(ctx context.Context, sessionID string) bool { if s == nil || sessionID == "" || sessionID == "ephemeral" { return false } @@ -1634,7 +1644,7 @@ type planStep struct { // live plan, not an archaeological record of every superseded generation. Pass // all=true for the audit/eval view that needs every generation (the // plan_generations assertion counts distinct generations across the full set). -func (s *store) getPlanSteps(ctx context.Context, sessionID string, all bool) ([]planStep, error) { +func (s *Store) GetPlanSteps(ctx context.Context, sessionID string, all bool) ([]planStep, error) { if s == nil { return nil, nil } @@ -1678,7 +1688,7 @@ type sessionQuestion struct { // getQuestions returns a task's questions (open and answered) newest-first — // REST hydration for the context panel's pinned question card and history. -func (s *store) getQuestions(ctx context.Context, sessionID string) ([]sessionQuestion, error) { +func (s *Store) GetQuestions(ctx context.Context, sessionID string) ([]sessionQuestion, error) { if s == nil { return nil, nil } @@ -1707,7 +1717,7 @@ func (s *store) getQuestions(ctx context.Context, sessionID string) ([]sessionQu // askOperator records a structured decision the agent needs from the operator, // moves the task to awaiting_input, and emits question.raised so the context // panel pins it. qctx carries {why, options, entities}. Returns the question id. -func (s *store) askOperator(ctx context.Context, sessionID, prompt string, qctx map[string]any) (string, error) { +func (s *Store) AskOperator(ctx context.Context, sessionID, prompt string, qctx map[string]any) (string, error) { if s == nil || sessionID == "" || sessionID == "ephemeral" { return "", nil } @@ -1730,7 +1740,7 @@ func (s *store) askOperator(ctx context.Context, sessionID, prompt string, qctx // openQuestionID returns the id of the session's open question, or "". Used to // auto-close a pending question when the operator answers via a plain chat reply. -func (s *store) openQuestionID(ctx context.Context, sessionID string) string { +func (s *Store) OpenQuestionID(ctx context.Context, sessionID string) string { if s == nil || sessionID == "" || sessionID == "ephemeral" { return "" } @@ -1742,7 +1752,7 @@ func (s *store) openQuestionID(ctx context.Context, sessionID string) string { // getQuestion returns a question's prompt, answer, and session — used to build // the resume note when the operator answers via the panel. -func (s *store) getQuestion(ctx context.Context, questionID string) (prompt, answer, sessionID string) { +func (s *Store) GetQuestion(ctx context.Context, questionID string) (prompt, answer, sessionID string) { if s == nil || questionID == "" { return "", "", "" } @@ -1759,7 +1769,7 @@ func (s *store) getQuestion(ctx context.Context, questionID string) (prompt, ans // and emits question.answered. It does NOT itself resume the agent — the caller // decides: a chat reply IS the resuming turn, while a panel answer triggers a // continuation. -func (s *store) answerQuestion(ctx context.Context, sessionID, questionID, answer string) error { +func (s *Store) AnswerQuestion(ctx context.Context, sessionID, questionID, answer string) error { if s == nil || sessionID == "" || sessionID == "ephemeral" || questionID == "" { return nil } @@ -1789,7 +1799,7 @@ var knowledgeSlugRe = regexp.MustCompile(`[a-z]+:nomos/[a-z0-9-]+`) // the entity it's "about" by upsert_knowledge itself — that about-link is the // retrieval path future tasks use (get_entity_knowledge); this task-link is for // the task's own outcome/knowledge view. -func (s *store) linkKnowledgeToTask(ctx context.Context, sessionID, resultText string) { +func (s *Store) LinkKnowledgeToTask(ctx context.Context, sessionID, resultText string) { if s == nil || sessionID == "" || sessionID == "ephemeral" { return } @@ -1817,7 +1827,7 @@ func (s *store) linkKnowledgeToTask(ctx context.Context, sessionID, resultText s map[string]any{"slug": slug}) } -func (s *store) updateSessionTitle(ctx context.Context, id, title string) error { +func (s *Store) UpdateSessionTitle(ctx context.Context, id, title string) error { if s == nil { return nil } @@ -1827,7 +1837,7 @@ func (s *store) updateSessionTitle(ctx context.Context, id, title string) error // resolveAgentID looks up the UUID of the agent entity (e.g. "agent:nomos"). // Returns uuid.Nil if the store is absent or the slug is unknown. -func (s *store) resolveAgentID(ctx context.Context, slug string) uuid.UUID { +func (s *Store) ResolveAgentID(ctx context.Context, slug string) uuid.UUID { if s == nil { return uuid.Nil } @@ -1839,10 +1849,10 @@ func (s *store) resolveAgentID(ctx context.Context, slug string) uuid.UUID { } // linkExecution records that a gated execution was initiated by a chat -// session, so the auto-continuation worker can feed its result back to that +// Session, so the auto-continuation worker can feed its result back to that // session when it finishes. Idempotent — the same execution may appear in // several tool results across a turn. -func (s *store) linkExecution(ctx context.Context, execID uuid.UUID, sessionID string) { +func (s *Store) LinkExecution(ctx context.Context, execID uuid.UUID, sessionID string) { if s == nil || execID == uuid.Nil || sessionID == "" || sessionID == "ephemeral" { return } @@ -1868,7 +1878,7 @@ var touchExcludedTypes = map[string]bool{"check": true, "execution": true, "task // 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) { +func (s *Store) RecordTouched(ctx context.Context, sessionID, toolName string, args map[string]any) { if s == nil || sessionID == "" || sessionID == "ephemeral" || len(args) == 0 { return } @@ -1953,7 +1963,7 @@ func collectTaskSlugs(v any, out map[string]struct{}) { // pendingContinuation is one finished execution whose result hasn't yet been // fed back to its originating session. -type pendingContinuation struct { +type PendingContinuation struct { ExecID uuid.UUID SessionID string Status string @@ -1964,7 +1974,7 @@ type pendingContinuation struct { // pendingContinuations returns executions that have reached a terminal state // but haven't been continued yet — the worker's work list. Bounded so one // tick can't fan out unboundedly. -func (s *store) pendingContinuations(ctx context.Context, limit int) []pendingContinuation { +func (s *Store) PendingContinuations(ctx context.Context, limit int) []PendingContinuation { if s == nil { return nil } @@ -1981,9 +1991,9 @@ func (s *store) pendingContinuations(ctx context.Context, limit int) []pendingCo return nil } defer rows.Close() - var out []pendingContinuation + var out []PendingContinuation for rows.Next() { - var p pendingContinuation + var p PendingContinuation if err := rows.Scan(&p.ExecID, &p.SessionID, &p.Status, &p.Result, &p.Action); err == nil { out = append(out, p) } @@ -1993,7 +2003,7 @@ func (s *store) pendingContinuations(ctx context.Context, limit int) []pendingCo // markContinued stamps an execution as fed-back so the worker won't process it // again (prevents an auto-continuation loop). -func (s *store) markContinued(ctx context.Context, execID uuid.UUID) { +func (s *Store) MarkContinued(ctx context.Context, execID uuid.UUID) { if s == nil { return } @@ -2001,29 +2011,29 @@ func (s *store) markContinued(ctx context.Context, execID uuid.UUID) { } // assentWindowActive reports whether THIS TASK currently has an open assent -// window — the scope gate for auto-continuation. Scoped by session, not just +// window — the scope gate for auto-continuation. Scoped by Session, not just // agent: with a single agent:nomos entity serving every concurrent task, an // agent-only key would let approving Task A's plan silently auto-run // unapproved config-mutation actions in a concurrently-running Task B. We // only auto-continue executions that are part of THIS session's approved // plan, never a stray action from another task riding the same window. -func (s *store) assentWindowActive(ctx context.Context, agentID uuid.UUID, sessionID string) bool { +func (s *Store) AssentWindowActive(ctx context.Context, agentID uuid.UUID, sessionID string) bool { if s == nil || agentID == uuid.Nil || sessionID == "" { return false // fail closed: no session to scope to means no window } var expires time.Time - key := assentWindowKey(agentID, sessionID) + key := AssentWindowKey(agentID, sessionID) if err := s.pool.QueryRow(ctx, `SELECT value::timestamptz FROM autonomy_settings WHERE key = $1`, key).Scan(&expires); err != nil { return false } return time.Now().Before(expires) } -// assentWindowKey scopes the grant to one agent AND one session/task — see +// AssentWindowKey scopes the grant to one agent AND one session/task — see // assentWindowActive. Must match internal/mcp/server.go's copy (mirrored // there, not shared, since the two are separate Go packages/binaries reading // the same autonomy_settings row). -func assentWindowKey(agentID uuid.UUID, sessionID string) string { +func AssentWindowKey(agentID uuid.UUID, sessionID string) string { return "assent_window.agent:" + agentID.String() + ".session:" + sessionID } @@ -2050,7 +2060,7 @@ func destructiveWindowKey(agentID uuid.UUID, targetSlug, sessionID string) strin // was gated independently. One explicit confirmation on a target should // cover the short follow-up sequence needed to finish what was just // confirmed — but only within the task that got the confirmation. -func (s *store) openDestructiveWindow(ctx context.Context, agentID uuid.UUID, targetSlug, sessionID string) { +func (s *Store) OpenDestructiveWindow(ctx context.Context, agentID uuid.UUID, targetSlug, sessionID string) { if s == nil || agentID == uuid.Nil || targetSlug == "" || sessionID == "" { return } @@ -2061,7 +2071,7 @@ func (s *store) openDestructiveWindow(ctx context.Context, agentID uuid.UUID, ta // destructiveWindowActive reports whether target has a live, explicitly- // confirmed destructive grant for this agent within this session/task. -func (s *store) destructiveWindowActive(ctx context.Context, agentID uuid.UUID, targetSlug, sessionID string) bool { +func (s *Store) DestructiveWindowActive(ctx context.Context, agentID uuid.UUID, targetSlug, sessionID string) bool { if s == nil || agentID == uuid.Nil || targetSlug == "" || sessionID == "" { return false } @@ -2076,7 +2086,7 @@ func (s *store) destructiveWindowActive(ctx context.Context, agentID uuid.UUID, // executionTarget resolves the target entity slug for an execution — used to // scope the destructive window to the right entity when a chat-assent typed // confirmation grants a destructive execution. -func (s *store) executionTarget(ctx context.Context, execID uuid.UUID) string { +func (s *Store) ExecutionTarget(ctx context.Context, execID uuid.UUID) string { if s == nil { return "" } @@ -2101,7 +2111,7 @@ var entityArgKeys = []string{ // resolveArgEntityID best-effort resolves the entity a tool call acted on // from its arguments, trying entityArgKeys in order. Returns uuid.Nil if no // key is present or none resolves to a known entity. -func (s *store) resolveArgEntityID(ctx context.Context, args map[string]any) uuid.UUID { +func (s *Store) ResolveArgEntityID(ctx context.Context, args map[string]any) uuid.UUID { if s == nil { return uuid.Nil } @@ -2126,11 +2136,11 @@ func (s *store) resolveArgEntityID(ctx context.Context, args map[string]any) uui // The (nullable) session_id column carries the conversation id. args is the // tool call's own arguments, used to best-effort tag the row with the // entity it acted on (see resolveArgEntityID). -func (s *store) logActivity(ctx context.Context, agentID uuid.UUID, sessionID, toolName string, args map[string]any, inputSummary, outputSummary string, durationMs int, success bool, correlationID string, tokenCount int) { +func (s *Store) LogActivity(ctx context.Context, agentID uuid.UUID, sessionID, toolName string, args map[string]any, inputSummary, outputSummary string, durationMs int, success bool, correlationID string, tokenCount int) { if s == nil || agentID == uuid.Nil { return } - entityID := s.resolveArgEntityID(ctx, args) + entityID := s.ResolveArgEntityID(ctx, args) var entityIDArg any if entityID != uuid.Nil { entityIDArg = entityID diff --git a/internal/nomos/session/store_test.go b/internal/nomos/session/store_test.go index bdaaf1c3..3e0a6f8f 100644 --- a/internal/nomos/session/store_test.go +++ b/internal/nomos/session/store_test.go @@ -1,4 +1,4 @@ -package main +package session // Integration tests against a real Postgres, mirroring // internal/db/integration_test.go's pattern: guarded by @@ -10,6 +10,7 @@ package main import ( "context" + "encoding/json" "errors" "fmt" "math/rand" @@ -98,7 +99,7 @@ func TestGetRecentMessages_Truncation(t *testing.T) { s := newTestStore(t) ctx := context.Background() - sess, err := s.createSession(ctx, "history window test") + sess, err := s.CreateSession(ctx, "history window test") if err != nil { t.Fatalf("createSession: %v", err) } @@ -111,12 +112,12 @@ func TestGetRecentMessages_Truncation(t *testing.T) { role = "assistant" } body := fmt.Appendf(nil, `{"role":%q,"text":"msg-%d"}`, role, i) - if err := s.saveMessage(ctx, sess.ID, role, body); err != nil { + if err := s.SaveMessage(ctx, sess.ID, role, body); err != nil { t.Fatalf("saveMessage %d: %v", i, err) } } - msgs, truncated, err := s.getRecentMessages(ctx, sess.ID, limit) + msgs, truncated, err := s.GetRecentMessages(ctx, sess.ID, limit) if err != nil { t.Fatalf("getRecentMessages: %v", err) } @@ -131,25 +132,25 @@ func TestGetRecentMessages_Truncation(t *testing.T) { // the last should be the most recently saved (msg-34). wantFirst := fmt.Sprintf("msg-%d", total-limit) wantLast := fmt.Sprintf("msg-%d", total-1) - if got := extractText(msgs[0].Content); got != wantFirst { + if got := sessionText(msgs[0].Content); got != wantFirst { t.Errorf("first retained message = %q, want %q", got, wantFirst) } - if got := extractText(msgs[len(msgs)-1].Content); got != wantLast { + if got := sessionText(msgs[len(msgs)-1].Content); got != wantLast { t.Errorf("last retained message = %q, want %q", got, wantLast) } // Under the limit: nothing dropped. - sess2, err := s.createSession(ctx, "small session") + sess2, err := s.CreateSession(ctx, "small session") if err != nil { t.Fatalf("createSession: %v", err) } for i := 0; i < 5; i++ { body := fmt.Appendf(nil, `{"role":"user","text":"msg-%d"}`, i) - if err := s.saveMessage(ctx, sess2.ID, "user", body); err != nil { + if err := s.SaveMessage(ctx, sess2.ID, "user", body); err != nil { t.Fatalf("saveMessage: %v", err) } } - msgs2, truncated2, err := s.getRecentMessages(ctx, sess2.ID, limit) + msgs2, truncated2, err := s.GetRecentMessages(ctx, sess2.ID, limit) if err != nil { t.Fatalf("getRecentMessages (small): %v", err) } @@ -173,14 +174,14 @@ func TestProposePlan_RefuseInFlight(t *testing.T) { s := newTestStore(t) ctx := context.Background() - sess, err := s.createSession(ctx, "plan refuse test") + sess, err := s.CreateSession(ctx, "plan refuse test") if err != nil { t.Fatalf("createSession: %v", err) } // First call: no steps exist yet — must persist as-is (replace mode, // trivially: nothing to replace). - out1, err := s.proposePlan(ctx, sess.ID, []PlanStepInput{{Title: "Step A"}}) + out1, err := s.ProposePlan(ctx, sess.ID, []PlanStepInput{{Title: "Step A"}}) if err != nil { t.Fatalf("proposePlan #1: %v", err) } @@ -192,20 +193,20 @@ func TestProposePlan_RefuseInFlight(t *testing.T) { } // Mark step 1 as started. - if err := s.updatePlanStep(ctx, sess.ID, 1, "running", "", ""); err != nil { + if err := s.UpdatePlanStep(ctx, sess.ID, 1, "running", "", ""); err != nil { t.Fatalf("updatePlanStep: %v", err) } // Second call, simulating a model that re-proposes mid-flight (the // operator-reported "proceed" bug): since step 1 has left 'pending', - // this MUST refuse with errPlanInFlight, not append or replace. - _, err = s.proposePlan(ctx, sess.ID, []PlanStepInput{{Title: "Step B"}}) - if !errors.Is(err, errPlanInFlight) { - t.Fatalf("proposePlan #2: err = %v, want errPlanInFlight (refuse mid-flight re-proposal)", err) + // this MUST refuse with ErrPlanInFlight, not append or replace. + _, err = s.ProposePlan(ctx, sess.ID, []PlanStepInput{{Title: "Step B"}}) + if !errors.Is(err, ErrPlanInFlight) { + t.Fatalf("proposePlan #2: err = %v, want ErrPlanInFlight (refuse mid-flight re-proposal)", err) } // The original step 1 must be untouched — not erased, not appended to. - steps, err := s.getPlanSteps(ctx, sess.ID, false) + steps, err := s.GetPlanSteps(ctx, sess.ID, false) if err != nil { t.Fatalf("getPlanSteps: %v", err) } @@ -219,18 +220,18 @@ func TestProposePlan_RefuseInFlight(t *testing.T) { // Third call BEFORE anything runs on a fresh session: every step is // still pending, so this must REPLACE (mark the prior plan `replaced`), // not refuse. The new plan becomes generation 2. - sess2, err := s.createSession(ctx, "plan replace test") + sess2, err := s.CreateSession(ctx, "plan replace test") if err != nil { t.Fatalf("createSession: %v", err) } - if _, err := s.proposePlan(ctx, sess2.ID, []PlanStepInput{{Title: "Original"}}); err != nil { + if _, err := s.ProposePlan(ctx, sess2.ID, []PlanStepInput{{Title: "Original"}}); err != nil { t.Fatalf("proposePlan (initial): %v", err) } - if _, err := s.proposePlan(ctx, sess2.ID, []PlanStepInput{{Title: "Revised"}}); err != nil { + if _, err := s.ProposePlan(ctx, sess2.ID, []PlanStepInput{{Title: "Revised"}}); err != nil { t.Fatalf("proposePlan (revise before execution): %v", err) } // Default (current generation) view: only the revised step. - revisedSteps, err := s.getPlanSteps(ctx, sess2.ID, false) + revisedSteps, err := s.GetPlanSteps(ctx, sess2.ID, false) if err != nil { t.Fatalf("getPlanSteps: %v", err) } @@ -244,7 +245,7 @@ func TestProposePlan_RefuseInFlight(t *testing.T) { t.Fatalf("revised step generation = %d, want 2 (prior pending plan is replaced, not deleted, so the counter increments)", revisedSteps[0].Generation) } // all=true audit view: both generations, the original marked `replaced`. - allSteps, err := s.getPlanSteps(ctx, sess2.ID, true) + allSteps, err := s.GetPlanSteps(ctx, sess2.ID, true) if err != nil { t.Fatalf("getPlanSteps(all): %v", err) } @@ -270,32 +271,32 @@ func TestUpdatePlanStep_GenerationRelative(t *testing.T) { s := newTestStore(t) ctx := context.Background() - sess, err := s.createSession(ctx, "gen-relative seq test") + sess, err := s.CreateSession(ctx, "gen-relative seq test") if err != nil { t.Fatalf("createSession: %v", err) } // Generation 1: two steps. - if _, err := s.proposePlan(ctx, sess.ID, []PlanStepInput{{Title: "A"}, {Title: "B"}}); err != nil { + if _, err := s.ProposePlan(ctx, sess.ID, []PlanStepInput{{Title: "A"}, {Title: "B"}}); err != nil { t.Fatalf("proposePlan #1: %v", err) } // Re-plan: setGoal marks the gen-1 plan `replaced`, proposePlan starts gen 2. - if err := s.setGoal(ctx, sess.ID, "follow-up sub-task"); err != nil { + if err := s.SetGoal(ctx, sess.ID, "follow-up sub-task"); err != nil { t.Fatalf("setGoal: %v", err) } - if _, err := s.proposePlan(ctx, sess.ID, []PlanStepInput{{Title: "C"}, {Title: "D"}}); err != nil { + if _, err := s.ProposePlan(ctx, sess.ID, []PlanStepInput{{Title: "C"}, {Title: "D"}}); err != nil { t.Fatalf("proposePlan #2: %v", err) } // The model addresses the new plan with 1-based seq. seq=1 must hit // gen-2 "C", leaving gen-1 "A" (replaced) untouched. - if err := s.updatePlanStep(ctx, sess.ID, 1, "running", "", ""); err != nil { + if err := s.UpdatePlanStep(ctx, sess.ID, 1, "running", "", ""); err != nil { t.Fatalf("updatePlanStep(seq=1, running): %v", err) } - if err := s.updatePlanStep(ctx, sess.ID, 1, "done", "", ""); err != nil { + if err := s.UpdatePlanStep(ctx, sess.ID, 1, "done", "", ""); err != nil { t.Fatalf("updatePlanStep(seq=1, done): %v", err) } - all, err := s.getPlanSteps(ctx, sess.ID, true) + all, err := s.GetPlanSteps(ctx, sess.ID, true) if err != nil { t.Fatalf("getPlanSteps(all): %v", err) } @@ -319,8 +320,8 @@ func TestUpdatePlanStep_GenerationRelative(t *testing.T) { } // Out-of-range seq must be refused (no current-gen step there). - if err := s.updatePlanStep(ctx, sess.ID, 99, "running", "", ""); !errors.Is(err, errPlanStepNotFound) { - t.Fatalf("updatePlanStep(seq=99) err = %v, want errPlanStepNotFound", err) + if err := s.UpdatePlanStep(ctx, sess.ID, 99, "running", "", ""); !errors.Is(err, ErrPlanStepNotFound) { + t.Fatalf("updatePlanStep(seq=99) err = %v, want ErrPlanStepNotFound", err) } } @@ -333,24 +334,24 @@ func TestCompleteTask_AutoCloseEmitsEvents(t *testing.T) { s := newTestStore(t) ctx := context.Background() - sess, err := s.createSession(ctx, "auto-close events test") + sess, err := s.CreateSession(ctx, "auto-close events test") if err != nil { t.Fatalf("createSession: %v", err) } - if _, err := s.proposePlan(ctx, sess.ID, []PlanStepInput{{Title: "A"}, {Title: "B"}}); err != nil { + if _, err := s.ProposePlan(ctx, sess.ID, []PlanStepInput{{Title: "A"}, {Title: "B"}}); err != nil { t.Fatalf("proposePlan: %v", err) } // A is running, B still pending at completion time. - if err := s.updatePlanStep(ctx, sess.ID, 1, "running", "", ""); err != nil { + if err := s.UpdatePlanStep(ctx, sess.ID, 1, "running", "", ""); err != nil { t.Fatalf("updatePlanStep(1, running): %v", err) } - if err := s.completeTask(ctx, sess.ID, "success", "done"); err != nil { + if err := s.CompleteTask(ctx, sess.ID, "success", "done"); err != nil { t.Fatalf("completeTask: %v", err) } // Every auto-closed step should now carry both a started_at and a // finished_at (no NULL-started `done` step). - steps, err := s.getPlanSteps(ctx, sess.ID, true) + steps, err := s.GetPlanSteps(ctx, sess.ID, true) if err != nil { t.Fatalf("getPlanSteps: %v", err) } @@ -382,64 +383,64 @@ func TestHadDiscoveryAndWriteback(t *testing.T) { s := newTestStore(t) ctx := context.Background() - sess, err := s.createSession(ctx, "discovery test") + sess, err := s.CreateSession(ctx, "discovery test") if err != nil { t.Fatalf("createSession: %v", err) } // Before any tool calls: no discovery, no writeback. - if s.hadDiscovery(ctx, sess.ID) { + if s.HadDiscovery(ctx, sess.ID) { t.Fatal("hadDiscovery = true before any tool calls, want false") } - if s.hadEntityWriteback(ctx, sess.ID) { + if s.HadEntityWriteback(ctx, sess.ID) { t.Fatal("hadEntityWriteback = true before any tool calls, want false") } // A `run` call (discovery) — should set hadDiscovery, not hadEntityWriteback. agentID := uuid.New() - s.logActivity(ctx, agentID, sess.ID, "run", nil, "", "uptime output", 100, true, "corr-1", 0) - if !s.hadDiscovery(ctx, sess.ID) { + s.LogActivity(ctx, agentID, sess.ID, "run", nil, "", "uptime output", 100, true, "corr-1", 0) + if !s.HadDiscovery(ctx, sess.ID) { t.Fatal("hadDiscovery = false after a successful run call, want true") } - if s.hadEntityWriteback(ctx, sess.ID) { + if s.HadEntityWriteback(ctx, sess.ID) { t.Fatal("hadEntityWriteback = true after only a run call, want false") } // A failed run call should NOT count as discovery (no facts learned). - sess2, err := s.createSession(ctx, "failed discovery test") + sess2, err := s.CreateSession(ctx, "failed discovery test") if err != nil { t.Fatalf("createSession: %v", err) } - s.logActivity(ctx, agentID, sess2.ID, "run", nil, "", "ssh timeout", 100, false, "corr-2", 0) - if s.hadDiscovery(ctx, sess2.ID) { + s.LogActivity(ctx, agentID, sess2.ID, "run", nil, "", "ssh timeout", 100, false, "corr-2", 0) + if s.HadDiscovery(ctx, sess2.ID) { t.Fatal("hadDiscovery = true after a failed run call, want false (no facts learned)") } // A get_entity call should NOT count as discovery (DB lookup, not live state). - sess3, err := s.createSession(ctx, "lookup test") + sess3, err := s.CreateSession(ctx, "lookup test") if err != nil { t.Fatalf("createSession: %v", err) } - s.logActivity(ctx, agentID, sess3.ID, "get_entity", nil, "", "entity row", 10, true, "corr-3", 0) - if s.hadDiscovery(ctx, sess3.ID) { + s.LogActivity(ctx, agentID, sess3.ID, "get_entity", nil, "", "entity row", 10, true, "corr-3", 0) + if s.HadDiscovery(ctx, sess3.ID) { t.Fatal("hadDiscovery = true after get_entity, want false (DB lookups are not discovery)") } // update_entity_attributes sets hadEntityWriteback. - sess4, err := s.createSession(ctx, "writeback test") + sess4, err := s.CreateSession(ctx, "writeback test") if err != nil { t.Fatalf("createSession: %v", err) } - s.logActivity(ctx, agentID, sess4.ID, "update_entity_attributes", nil, "", "ok", 10, true, "corr-4", 0) - if !s.hadEntityWriteback(ctx, sess4.ID) { + s.LogActivity(ctx, agentID, sess4.ID, "update_entity_attributes", nil, "", "ok", 10, true, "corr-4", 0) + if !s.HadEntityWriteback(ctx, sess4.ID) { t.Fatal("hadEntityWriteback = false after update_entity_attributes, want true") } // And the discovery+writeback combination (the conv3 scenario). - s.logActivity(ctx, agentID, sess4.ID, "run", nil, "", "apt-get update output", 100, true, "corr-5", 0) - if !s.hadDiscovery(ctx, sess4.ID) { + s.LogActivity(ctx, agentID, sess4.ID, "run", nil, "", "apt-get update output", 100, true, "corr-5", 0) + if !s.HadDiscovery(ctx, sess4.ID) { t.Fatal("hadDiscovery = false after run+writeback, want true") } - if !s.hadEntityWriteback(ctx, sess4.ID) { + if !s.HadEntityWriteback(ctx, sess4.ID) { t.Fatal("hadEntityWriteback = false after run+writeback, want true") } } @@ -459,13 +460,13 @@ func TestSetGoal_SupersededEvent(t *testing.T) { s := newTestStore(t) ctx := context.Background() - sess, err := s.createSession(ctx, "goal pivot test") + sess, err := s.CreateSession(ctx, "goal pivot test") if err != nil { t.Fatalf("createSession: %v", err) } // First set_goal — no prior, no supersession event expected. - if err := s.setGoal(ctx, sess.ID, "Fix sabnzbd download folder to use ludo-lvm"); err != nil { + if err := s.SetGoal(ctx, sess.ID, "Fix sabnzbd download folder to use ludo-lvm"); err != nil { t.Fatalf("setGoal #1: %v", err) } if n := countEvents(ctx, s, sess.ID, "task.superseded"); n != 0 { @@ -473,7 +474,7 @@ func TestSetGoal_SupersededEvent(t *testing.T) { } // Second set_goal with a DIFFERENT goal — supersession event expected. - if err := s.setGoal(ctx, sess.ID, "Add NFS export of ludo-lvm to ZimaOS"); err != nil { + if err := s.SetGoal(ctx, sess.ID, "Add NFS export of ludo-lvm to ZimaOS"); err != nil { t.Fatalf("setGoal #2: %v", err) } if n := countEvents(ctx, s, sess.ID, "task.superseded"); n != 1 { @@ -482,7 +483,7 @@ func TestSetGoal_SupersededEvent(t *testing.T) { // Third set_goal with the SAME goal as the second — no new supersession // event (idempotent: same goal is a no-op, not a pivot). - if err := s.setGoal(ctx, sess.ID, "Add NFS export of ludo-lvm to ZimaOS"); err != nil { + if err := s.SetGoal(ctx, sess.ID, "Add NFS export of ludo-lvm to ZimaOS"); err != nil { t.Fatalf("setGoal #3: %v", err) } if n := countEvents(ctx, s, sess.ID, "task.superseded"); n != 1 { @@ -490,7 +491,7 @@ func TestSetGoal_SupersededEvent(t *testing.T) { } // The session's current goal must be the latest one set. - got, err := s.getSession(ctx, sess.ID) + got, err := s.GetSession(ctx, sess.ID) if err != nil { t.Fatalf("getSession: %v", err) } @@ -509,3 +510,14 @@ func countEvents(ctx context.Context, s *Store, sessionID, eventType string) int sessionID, eventType).Scan(&n) return n } + +// sessionText pulls the "text" field from a persisted message's JSONB content. +func sessionText(content json.RawMessage) string { + var m struct { + Text string `json:"text"` + } + if err := json.Unmarshal(content, &m); err != nil { + return "" + } + return m.Text +} diff --git a/plans/index.md b/plans/index.md index e65a2c1f..4a1f7698 100644 --- a/plans/index.md +++ b/plans/index.md @@ -6,7 +6,9 @@ went sideways, open an investigation. ## Active -All plans reviewed and moved to `done/` — see the [Done table](#done) below. +| Date | Title | +| ---- | ----- | +| 2026-08-16 | [dsh-as-agent: replace nomos with DeepSeek Harness](2026-08-16-dsh-as-agent-replace-nomos.md) | ## Done