diff --git a/cmd/nomos/store.go b/cmd/nomos/store.go index 8c7f0d0..b23612d 100644 --- a/cmd/nomos/store.go +++ b/cmd/nomos/store.go @@ -265,6 +265,138 @@ 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 +// 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 { + 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 { + return nil + } + return &id +} + +// setGoal records the task's goal and moves it into planning. Emits goal.set. +func (s *store) setGoal(ctx context.Context, sessionID, goal string) error { + if s == nil || sessionID == "" || sessionID == "ephemeral" { + return nil + } + if _, err := s.pool.Exec(ctx, + `UPDATE agent_sessions SET goal = $2, status = 'planning', last_active_at = now() WHERE id = $1`, + sessionID, goal); err != nil { + return err + } + _ = observability.Event(ctx, sqlcgen.New(s.pool), "goal.set", s.taskEntityPtr(ctx, sessionID), + "info", "nomos", sessionID, map[string]any{"goal": goal}) + return nil +} + +// planStepInput is one step as the agent proposes it. +type planStepInput struct { + Title string + Detail string + TargetSlug string +} + +// proposePlan replaces the task's plan with a fresh ordered step list and moves +// the task into executing. v1 does a clean replace (delete + insert): revising a +// plan mid-flight starts a new list rather than versioning the old one. Emits +// plan.proposed with the persisted steps (seq + id) so the panel can render and +// later address them by id. +func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planStepInput) ([]map[string]any, error) { + if s == nil || sessionID == "" || sessionID == "ephemeral" { + return nil, nil + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + if _, err := tx.Exec(ctx, `DELETE FROM session_plan_steps WHERE session_id = $1`, sessionID); err != nil { + return nil, err + } + out := make([]map[string]any, 0, len(steps)) + for i, st := range steps { + var targetSlug *string + if st.TargetSlug != "" { + targetSlug = &st.TargetSlug + } + var id uuid.UUID + if err := tx.QueryRow(ctx, ` + INSERT INTO session_plan_steps (session_id, seq, title, detail, target_slug) + VALUES ($1, $2, $3, $4, $5) RETURNING id`, + sessionID, i+1, st.Title, st.Detail, targetSlug).Scan(&id); err != nil { + return nil, err + } + out = append(out, map[string]any{ + "id": id.String(), "seq": i + 1, "title": st.Title, + "detail": st.Detail, "target_slug": st.TargetSlug, + }) + } + if _, err := tx.Exec(ctx, + `UPDATE agent_sessions SET status = 'executing', last_active_at = now() WHERE id = $1`, sessionID); err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + // Event after commit so subscribers only ever see a persisted plan. + _ = observability.Event(ctx, sqlcgen.New(s.pool), "plan.proposed", s.taskEntityPtr(ctx, sessionID), + "info", "nomos", sessionID, map[string]any{"steps": out}) + return out, nil +} + +// updatePlanStep sets a step's status by seq, stamping started_at/finished_at +// and linking an execution if given. Emits plan.step.started (running) or +// plan.step.finished (terminal) so the panel advances live. The execution link +// is also what lets the api auto-close the step when the execution finishes +// (see closePlanStepForExecution). +func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, status, execID string) error { + if s == nil || sessionID == "" || sessionID == "ephemeral" { + return nil + } + stamp := "" + switch status { + case "running": + stamp = ", started_at = COALESCE(started_at, now())" + case "done", "failed", "skipped", "blocked": + stamp = ", finished_at = now()" + } + var execPtr *uuid.UUID + if id, err := uuid.Parse(execID); err == nil { + execPtr = &id + } + var stepID uuid.UUID + var targetSlug *string + // stamp is a fixed literal from the switch above — never user input. + if err := s.pool.QueryRow(ctx, ` + UPDATE session_plan_steps + SET status = $3, execution_id = COALESCE($4, execution_id)`+stamp+` + WHERE session_id = $1 AND seq = $2 + RETURNING id, target_slug`, sessionID, seq, status, execPtr).Scan(&stepID, &targetSlug); err != nil { + return err + } + // Anchor the event to the step's target entity when it has one, else the task. + entPtr := s.taskEntityPtr(ctx, sessionID) + if targetSlug != nil && *targetSlug != "" { + var tid uuid.UUID + if s.pool.QueryRow(ctx, `SELECT id FROM entities WHERE slug = $1`, *targetSlug).Scan(&tid) == nil { + entPtr = &tid + } + } + evType := "plan.step.finished" + if status == "running" { + evType = "plan.step.started" + } + data := map[string]any{"step_id": stepID.String(), "seq": seq, "status": status} + if execID != "" { + data["execution_id"] = execID + } + _ = observability.Event(ctx, sqlcgen.New(s.pool), evType, entPtr, "info", "nomos", sessionID, data) + return nil +} + // 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 diff --git a/cmd/nomos/tasks.go b/cmd/nomos/tasks.go index 53c130d..23bec92 100644 --- a/cmd/nomos/tasks.go +++ b/cmd/nomos/tasks.go @@ -3,6 +3,7 @@ package main import ( "context" "fmt" + "strings" ) // Task tools are nomos-LOCAL, not MCP tools. They are session-scoped, and the @@ -16,6 +17,64 @@ import ( func taskToolDefs() []toolDef { return []toolDef{ + { + Name: "set_goal", + Description: "State the goal of this task in one sentence, as early as you " + + "can. This is what the task is trying to achieve (e.g. 'Deploy TypeType " + + "as an LXC on strong'); it heads the task on the board and the context " + + "panel. Call it once you understand what the operator wants.", + InputSchema: map[string]any{ + "type": "object", + "properties": map[string]any{ + "goal": map[string]any{"type": "string", "description": "The task's goal, one sentence."}, + }, + "required": []string{"goal"}, + }, + }, + { + Name: "propose_plan", + Description: "Lay out the ordered steps you'll take to reach the goal. The " + + "operator sees these in the context panel and watches them progress. " + + "Call this before you start executing (after gathering what you need); " + + "re-call it to revise the plan. As you work, call update_plan_step to " + + "advance each one.", + InputSchema: map[string]any{ + "type": "object", + "properties": map[string]any{ + "steps": map[string]any{ + "type": "array", + "description": "Ordered steps, first to last.", + "items": map[string]any{ + "type": "object", + "properties": map[string]any{ + "title": map[string]any{"type": "string", "description": "Short imperative step title (e.g. 'Create the LXC')."}, + "detail": map[string]any{"type": "string", "description": "Optional one-line detail."}, + "target_slug": map[string]any{"type": "string", "description": "Optional entity slug this step acts on (e.g. lxc:typetype)."}, + }, + "required": []string{"title"}, + }, + }, + }, + "required": []string{"steps"}, + }, + }, + { + Name: "update_plan_step", + Description: "Advance a plan step as you work it. Set status to 'running' when " + + "you start it (pass execution_id if the step queued a gated action, so " + + "the board can auto-close it when that finishes), then 'done' / 'failed' " + + "/ 'skipped' / 'blocked' when it resolves. Keeps the operator's progress " + + "view honest.", + InputSchema: map[string]any{ + "type": "object", + "properties": map[string]any{ + "seq": map[string]any{"type": "integer", "description": "1-based step number from propose_plan."}, + "status": map[string]any{"type": "string", "enum": []string{"running", "done", "failed", "skipped", "blocked"}, "description": "New status for the step."}, + "execution_id": map[string]any{"type": "string", "description": "Optional execution UUID this step is running, so it auto-closes on completion."}, + }, + "required": []string{"seq", "status"}, + }, + }, { Name: "complete_task", Description: "Mark the current task finished. Call this once the goal is " + @@ -44,18 +103,77 @@ func taskToolDefs() []toolDef { func isTaskTool(name string) bool { switch name { - case "complete_task": + case "set_goal", "propose_plan", "update_plan_step", "complete_task": return true default: return false } } +// toInt coerces a JSON tool-arg number (float64 after unmarshal) to int. +func toInt(v any) int { + switch n := v.(type) { + case float64: + return int(n) + case int: + return n + default: + return 0 + } +} + // handleTaskTool executes a nomos-local task tool. Returns (result, true) if it // handled the call, or (nil, false) if name is not a local task tool (so the // caller forwards it to the MCP client). func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args map[string]any) (any, bool) { switch name { + case "set_goal": + goal, _ := args["goal"].(string) + if strings.TrimSpace(goal) == "" { + return "error: set_goal needs a goal", true + } + if err := a.store.setGoal(ctx, sessionID, goal); err != nil { + return fmt.Sprintf("error setting goal: %v", err), true + } + return "Goal set: " + goal, true + + case "propose_plan": + raw, _ := args["steps"].([]any) + var steps []planStepInput + for _, r := range raw { + m, ok := r.(map[string]any) + if !ok { + continue + } + title, _ := m["title"].(string) + if strings.TrimSpace(title) == "" { + continue + } + detail, _ := m["detail"].(string) + target, _ := m["target_slug"].(string) + steps = append(steps, planStepInput{Title: title, Detail: detail, TargetSlug: target}) + } + if len(steps) == 0 { + return "error: propose_plan needs at least one step with a title", true + } + persisted, err := a.store.proposePlan(ctx, sessionID, steps) + if err != nil { + return fmt.Sprintf("error proposing plan: %v", err), true + } + return fmt.Sprintf("Plan set: %d step(s). Execute them now, marking each with update_plan_step as you go.", len(persisted)), true + + case "update_plan_step": + seq := toInt(args["seq"]) + status, _ := args["status"].(string) + execID, _ := args["execution_id"].(string) + if seq <= 0 || status == "" { + return "error: update_plan_step needs seq (>=1) and status", true + } + if err := a.store.updatePlanStep(ctx, sessionID, seq, status, execID); err != nil { + return fmt.Sprintf("error updating step %d: %v", seq, err), true + } + return fmt.Sprintf("Step %d → %s", seq, status), true + case "complete_task": outcome, _ := args["outcome"].(string) summary, _ := args["summary"].(string) diff --git a/internal/httpapi/phase3.go b/internal/httpapi/phase3.go index 08637bc..2e0de96 100644 --- a/internal/httpapi/phase3.go +++ b/internal/httpapi/phase3.go @@ -231,6 +231,32 @@ func emitExecutionEvent(ctx context.Context, pool *db.Pool, execID uuid.UUID, st severity = "warning" } _ = observability.Event(ctx, sqlcgen.New(pool), "execution."+status, &execID, severity, "actuator", "", detail) + if status == "completed" || status == "failed" || status == "cancelled" { + closePlanStepForExecution(ctx, pool, execID, status) + } +} + +// closePlanStepForExecution auto-closes a task plan step whose linked execution +// just reached a terminal state, so the task board advances even if the agent +// doesn't call update_plan_step itself (belt and suspenders — the agent links +// the step to the execution when it starts it; the api finishes it here). Emits +// plan.step.finished correlated to the step's session. No-op for the vast +// majority of executions, which aren't plan steps. +func closePlanStepForExecution(ctx context.Context, pool *db.Pool, execID uuid.UUID, execStatus string) { + stepStatus := "done" + if execStatus == "failed" || execStatus == "cancelled" { + stepStatus = "failed" + } + var stepID, sessionID string + var seq int + if err := pool.QueryRow(ctx, ` + UPDATE session_plan_steps SET status = $2, finished_at = now() + WHERE execution_id = $1 AND status NOT IN ('done', 'failed', 'skipped') + RETURNING id::text, session_id::text, seq`, execID, stepStatus).Scan(&stepID, &sessionID, &seq); err != nil { + return // no matching open step + } + _ = observability.Event(ctx, sqlcgen.New(pool), "plan.step.finished", &execID, "info", "actuator", sessionID, + map[string]any{"step_id": stepID, "seq": seq, "status": stepStatus, "execution_id": execID.String()}) } func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID, targetSlug string, actionStr string) {