session reliability: reconnect, knowledge loop, retire request_execution
Some checks failed
ci / build-test (push) Has been cancelled
ci / docker-build (push) Has been cancelled
Desktop App / Build Linux (amd64) (push) Has been cancelled
Desktop App / Attach to Release (push) Has been cancelled

Phase 1 — crash recovery: SSE auto-reconnect + backoff, polling gate
during disconnect, connection banner with retry button, empty-response
retry 3x, non-terminal resume on empty response, persistent error cards.

Phase 2/4 — visibility + continuation: custom ExecutionStatus renderer,
approvals extracted on every tool_result (not just done), activity bar
with status/goal, SessionDigest live polling, Continue button.

Phase 3 — cleanup: complete_task auto-cancels orphaned approvals,
deletes assent/destructive window keys, propose_plan marks pending
steps as replaced, plan step seq-order enforcement.

Phase 5 — knowledge loop: list_lxcs state filter (active/destroyed),
SOUL.md unmissable writeback section, propose_plan validation nudge,
complete_task writeback check, upsert_knowledge about array support,
plan generation grouping in frontend, session approval count badge.

Retire request_execution — all mutations now route through run.
Updated SOUL.md, AGENTS.md, CLIENTS.md, skills, and agent system notes.

Migration 020: plan step generation column, audit_log session_id index,
nomos_plan_executions pending-approval index.
This commit is contained in:
2026-07-14 11:03:23 +02:00
parent b446909ea5
commit 60effcb2fe
25 changed files with 1894 additions and 323 deletions

View File

@@ -78,7 +78,7 @@ the REST API. Closest current equivalents for what used to live here:
| `homelab signal list\|ack\|resolve\|mute` | MCP `get_signal_history`, or REST `POST /api/v1/signals/{id}/ack\|resolve\|mute` (the control-room UI's Signals page wraps these) | | `homelab signal list\|ack\|resolve\|mute` | MCP `get_signal_history`, or REST `POST /api/v1/signals/{id}/ack\|resolve\|mute` (the control-room UI's Signals page wraps these) |
| `homelab approval request\|list\|reply\|check` | REST `GET/POST /api/v1/approvals*` (Matrix-delivered via the notifier, or the control-room UI's Operations page) | | `homelab approval request\|list\|reply\|check` | REST `GET/POST /api/v1/approvals*` (Matrix-delivered via the notifier, or the control-room UI's Operations page) |
| `homelab restart <service> --approval-id <id>` | MCP `run` (policy-gated — auto-executes if read-only/reversible_low, otherwise queues for the same Matrix/UI approval) | | `homelab restart <service> --approval-id <id>` | MCP `run` (policy-gated — auto-executes if read-only/reversible_low, otherwise queues for the same Matrix/UI approval) |
| `homelab decide <action> <entity>` | No direct equivalent — classification now happens inline inside `run`/`request_execution`, not as a separate dry-run call | | `homelab decide <action> <entity>` | No direct equivalent — classification now happens inline inside `run`, not as a separate dry-run call |
There is no separately-deployed "Oikos Console" anymore — the control-room There is no separately-deployed "Oikos Console" anymore — the control-room
SPA (`web/`) is the operator dashboard, served standalone (see SPA (`web/`) is the operator dashboard, served standalone (see

View File

@@ -36,7 +36,7 @@ For each session determine:
| Signature | Root cause | Fix | | Signature | Root cause | Fix |
|-----------|-----------|-----| |-----------|-----------|-----|
| Agent: "I can't run X — only supports Y" | Missing action in `request_execution` | Add action in `internal/mcp/server.go` | | Agent: "I can't run X" | Missing target or capability | Use `run` with shell command — there is no fixed action enum anymore |
| Agent: "No local knowledge on that" + no web tool | Missing `http_get` / web fetch MCP tool | Add MCP tool | | Agent: "No local knowledge on that" + no web tool | Missing `http_get` / web fetch MCP tool | Add MCP tool |
| Empty assistant bubble (text="", no tools) | Model returned blank completion | Retry + error surfacing | | Empty assistant bubble (text="", no tools) | Model returned blank completion | Retry + error surfacing |
| Non-English boilerplate refusal | Flash-tier model degradation | Response quality guard | | Non-English boilerplate refusal | Flash-tier model degradation | Response quality guard |
@@ -75,7 +75,7 @@ Session: {id[:8]} — "{title[:60]}"
- `cmd/nomos/agent.go` — agent loop, tool building, response guards - `cmd/nomos/agent.go` — agent loop, tool building, response guards
- `cmd/nomos/store.go` — session + message persistence - `cmd/nomos/store.go` — session + message persistence
- `internal/mcp/server.go` — all tool implementations including `request_execution` - `internal/mcp/server.go` — all tool implementations (`run`, `list_lxcs`, …)
- `web/src/lib/components/ToolCallGroup.svelte` — tool result display - `web/src/lib/components/ToolCallGroup.svelte` — tool result display
- `nomos/SOUL.md` — agent persona and tool selection rules - `nomos/SOUL.md` — agent persona and tool selection rules
- `plans/2026-07-09-chat-sessions-improvements.md` — prior session findings - `plans/2026-07-09-chat-sessions-improvements.md` — prior session findings

1
.gitignore vendored
View File

@@ -22,3 +22,4 @@ web/node_modules/
cmd/desktop/frontend/dist/ cmd/desktop/frontend/dist/
cmd/desktop/build/ cmd/desktop/build/
cmd/desktop/Oikos cmd/desktop/Oikos
desktop

View File

@@ -110,9 +110,9 @@ Available tools (33 total):
read-only inspection runs immediately, anything state-changing needs read-only inspection runs immediately, anything state-changing needs
operator approval, and destructive patterns (rm -rf, dd, mkfs, operator approval, and destructive patterns (rm -rf, dd, mkfs,
pct/qm destroy, DROP TABLE, reboot, curl-pipe-to-shell, ...) always pct/qm destroy, DROP TABLE, reboot, curl-pipe-to-shell, ...) always
need approval regardless of what you declare. Prefer this over need approval regardless of what you declare. This is the ONLY
request_execution for anything not already covered by its fixed enum. mutation tool — `request_execution` was retired 2026-07-14.
request_execution(target, action, params) — the older, fixed-enum path `run` — the general execution primitive. Run any shell
(restart, systemctl, pct_exec, apt_upgrade, pct_create). Still the (restart, systemctl, pct_exec, apt_upgrade, pct_create). Still the
route for those specific actions; policy-gated the same way `run` is. route for those specific actions; policy-gated the same way `run` is.
get_execution_status(execution_id) — poll progress get_execution_status(execution_id) — poll progress
@@ -165,8 +165,7 @@ per the DB-as-source-of-truth plan.
operator interface — it has 33 MCP tools for observe/orient/decide/act operator interface — it has 33 MCP tools for observe/orient/decide/act
(§3). (§3).
- **Actions** (restart, logs, apt, pct exec, or anything else): Nomos calls - **Actions** (restart, logs, apt, pct exec, or anything else): Nomos calls
`run` (the general execution primitive) or `request_execution` (the older `run` (the general execution primitive) via MCP. `reversible_low`/read-only actions execute
fixed-enum path) via MCP. `reversible_low`/read-only actions execute
immediately; `config_mutation` and `destructive` actions are queued for immediately; `config_mutation` and `destructive` actions are queued for
operator approval via Matrix or the control-room UI's Operations page. operator approval via Matrix or the control-room UI's Operations page.
- **Secrets**: managed by Infisical (`oikos secret` subcommand for - **Secrets**: managed by Infisical (`oikos secret` subcommand for

View File

@@ -28,7 +28,7 @@ Docker stack on mac-mini and exposes an MCP server + REST API.
| Record a discovered fact/relationship | MCP `update_entity_attributes`, `create_relationship`, `upsert_knowledge` | | Record a discovered fact/relationship | MCP `update_entity_attributes`, `create_relationship`, `upsert_knowledge` |
Most MCP tools are read-only; a few mutate the knowledge graph (recording Most MCP tools are read-only; a few mutate the knowledge graph (recording
what you learned) or the live infrastructure (`run`, `request_execution`), what you learned) or the live infrastructure (`run`),
gated by risk classification and — for `config_mutation`/`destructive` gated by risk classification and — for `config_mutation`/`destructive`
actions — operator approval. See [AGENTS.md](AGENTS.md#3-the-mcp-server) for actions — operator approval. See [AGENTS.md](AGENTS.md#3-the-mcp-server) for
the full tool catalog. the full tool catalog.

View File

@@ -17,12 +17,12 @@ import (
) )
// maxIterations bounds one chat turn's tool-calling loop. Provisioning a // maxIterations bounds one chat turn's tool-calling loop. Provisioning a
// service is a long chain (research → plan → request_execution → per-step // service is a long chain (research → plan → run → per-step
// install/verify run calls), so this must be generous; a full deploy with the // install/verify run calls), so this must be generous; a full deploy with the
// decomposed pct_create flow can legitimately need many steps. On exhaustion // decomposed pct_create flow can legitimately need many steps. On exhaustion
// the loop now produces a real summary (finalSummary) rather than a dead end. // the loop now produces a real summary (finalSummary) rather than a dead end.
const maxIterations = 40 const maxIterations = 40
const maxLLMRetries = 1 const maxLLMRetries = 2
// historyWindowSize bounds how many of a session's most recent persisted // historyWindowSize bounds how many of a session's most recent persisted
// messages are replayed into the LLM's context on each turn — see // messages are replayed into the LLM's context on each turn — see
@@ -129,7 +129,7 @@ func loadSoul() string {
} }
return `You are Nomos, the steward of the oikos — the AI agent for the hubris homelab. return `You are Nomos, the steward of the oikos — the AI agent for the hubris homelab.
You have access to MCP tools to query topology, health, knowledge, and request You have access to MCP tools to query topology, health, knowledge, and request
gated mutations through request_execution. Be concise. Prefer tools over guessing.` gated mutations through run. Be concise. Prefer tools over guessing.`
} }
// assentWindowDuration is how long after an operator approves a plan that // assentWindowDuration is how long after an operator approves a plan that
@@ -294,7 +294,7 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s
} }
if len(granted) > 0 { if len(granted) > 0 {
a.openAssentWindow(ctx, sessionID) a.openAssentWindow(ctx, sessionID)
note := fmt.Sprintf("[System: the operator's message approved pending execution(s) %s via chat assent — they are now running. An assent window is now active for 30 minutes: config_mutation commands will auto-run without re-approval. Do not re-request or call request_execution/run again for these; check get_execution_status if you need the outcome. CONTINUE executing the full plan — do not stop and wait for 'continue' after each step. Only surface to the operator for destructive actions (need typed confirmation) or if you're genuinely stuck after trying alternatives.]", strings.Join(granted, ", ")) note := fmt.Sprintf("[System: the operator's message approved pending execution(s) %s via chat assent — they are now running. An assent window is now active for 30 minutes: config_mutation commands will auto-run without re-approval. Do not re-request or call run again for these; check get_execution_status if you need the outcome. CONTINUE executing the full plan — do not stop and wait for 'continue' after each step. Only surface to the operator for destructive actions (need typed confirmation) or if you're genuinely stuck after trying alternatives.]", strings.Join(granted, ", "))
messages = append(messages, openai.SystemMessage(note)) messages = append(messages, openai.SystemMessage(note))
} }
if len(blocked) > 0 { if len(blocked) > 0 {
@@ -305,9 +305,9 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s
// The operator said "proceed"/"go ahead"/"yes" but the preceding // The operator said "proceed"/"go ahead"/"yes" but the preceding
// assistant turn had NO pending approvals — meaning the agent // assistant turn had NO pending approvals — meaning the agent
// proposed a plan in text and asked "shall I?" without calling // proposed a plan in text and asked "shall I?" without calling
// request_execution yet. Inject a system note telling the agent // run yet. Inject a system note telling the agent
// the operator approved — go execute the plan now. // the operator approved — go execute the plan now.
note := "[System: The operator approved your proposed plan. Execute it now — call request_execution or run to carry out the steps you described. Do not re-describe the plan or ask for confirmation again. The assent window is active: config_mutation commands will auto-run once you create them.]" note := "[System: The operator approved your proposed plan. Execute it now — call run to carry out the steps you described. Do not re-describe the plan or ask for confirmation again. The assent window is active: config_mutation commands will auto-run once you create them.]"
messages = append(messages, openai.SystemMessage(note)) messages = append(messages, openai.SystemMessage(note))
a.openAssentWindow(ctx, sessionID) a.openAssentWindow(ctx, sessionID)
} }

View File

@@ -52,7 +52,7 @@ func (a *agent) runIdleSweepWorker(ctx context.Context) {
return return
} }
slog.Info("nomos: idle sweep worker started") slog.Info("nomos: idle sweep worker started")
ticker := time.NewTicker(5 * time.Minute) ticker := time.NewTicker(2 * time.Minute)
defer ticker.Stop() defer ticker.Stop()
for { for {
select { select {
@@ -143,12 +143,25 @@ func (a *agent) processContinuations(ctx context.Context) {
// multiple tasks in flight, one task's open window must never cover a // multiple tasks in flight, one task's open window must never cover a
// pending continuation belonging to a different task. // pending continuation belonging to a different task.
if !a.store.assentWindowActive(ctx, a.agentID, p.SessionID) { if !a.store.assentWindowActive(ctx, a.agentID, p.SessionID) {
// A finished one-off execution with no window is left as-is // Re-open the assent window if this session is genuinely
// (marked continued so we don't re-check it forever) — the // executing (plan was approved, work is in progress) — the
// operator decides what happens next, as today. // window may have expired while the execution ran. Don't
// penalize timing: the plan was approved, the work happened,
// the result should flow back.
sesh, seshErr := a.store.getSession(ctx, p.SessionID)
if seshErr == nil && sesh.Goal != "" && (sesh.Status == "executing" || sesh.Status == "planning") {
a.openAssentWindow(ctx, p.SessionID)
slog.Info("nomos: re-opened assent window for continuing session", "session", p.SessionID, "execution", p.ExecID)
} else {
// Genuinely no plan — inject a visible note so the
// operator knows WHY the agent didn't auto-continue.
note := fmt.Sprintf("[System: execution %s finished with status=%s, but the assent window for this session is not active. The agent will not auto-continue. Reply 'continue' or re-approve the plan to resume.]", p.ExecID, p.Status)
body, _ := json.Marshal(map[string]any{"role": "assistant", "text": note, "auto": true})
a.store.saveMessage(context.Background(), p.SessionID, "assistant", body)
a.store.markContinued(ctx, p.ExecID) a.store.markContinued(ctx, p.ExecID)
continue continue
} }
}
a.store.markContinued(ctx, p.ExecID) // stamp first: a failure here must not cause a re-continue loop a.store.markContinued(ctx, p.ExecID) // stamp first: a failure here must not cause a re-continue loop
safego.Go("nomos:continue-session:"+p.SessionID, func() { a.continueSession(ctx, p) }) safego.Go("nomos:continue-session:"+p.SessionID, func() { a.continueSession(ctx, p) })
} }
@@ -213,7 +226,7 @@ func (a *agent) resumeSession(ctx context.Context, sessionID, note string) {
// without this outer retry the operator would see nothing at all. // without this outer retry the operator would see nothing at all.
cctx, cancel := context.WithTimeout(ctx, 10*time.Minute) cctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
defer cancel() defer cancel()
for attempt := 0; attempt < 2; attempt++ { for attempt := 0; attempt < 3; attempt++ {
toolCalls, finalText, errText = nil, "", "" toolCalls, finalText, errText = nil, "", ""
emit := func(ev agentEvent) { emit := func(ev agentEvent) {
if ev.Type == "tool_use" || ev.Type == "tool_result" { if ev.Type == "tool_use" || ev.Type == "tool_result" {
@@ -241,21 +254,24 @@ func (a *agent) resumeSession(ctx context.Context, sessionID, note string) {
if errText != "" && finalText == "" { if errText != "" && finalText == "" {
slog.Error("nomos: resume produced no response after retry", "session", sessionID, "error", errText) slog.Error("nomos: resume produced no response after retry", "session", sessionID, "error", errText)
// Give the task a real, operator-visible terminal state instead of // Persist a visible system note in the transcript so the
// leaving it silently stuck at whatever status it was in (typically // operator sees what happened, but do NOT auto-complete the
// 'executing' or 'awaiting_input') forever. Before this, a // task — leave it in 'executing' so a follow-up chat message
// permanently-failed resume was invisible beyond a log line — the // can resume it. Before this fix, the task was marked 'failed'
// task board just showed a task that never changed, with nothing // here, which ended it permanently and required starting over.
// telling the operator it needed attention. Marking it failed here resumeFailedNote := fmt.Sprintf("[System: auto-resume failed after retrying: %s. The task is paused — send another message to continue.]", errText)
// doesn't prevent the operator from continuing to work the task via body, _ := json.Marshal(map[string]any{
// a fresh chat message afterward; it just stops the silent hang. "role": "assistant",
summary := fmt.Sprintf("Auto-resume failed after retrying: %s", errText) "text": resumeFailedNote,
if len(summary) > 200 { "auto": true,
summary = summary[:200] + "…" })
} if msgID != uuid.Nil {
if cerr := a.store.completeTask(context.Background(), sessionID, "failure", summary); cerr != nil { a.store.updateMessage(context.Background(), msgID, body)
slog.Error("nomos: failed to mark task failed after resume gave up", "session", sessionID, "error", cerr) } else {
// No placeholder was inserted (rare), save directly.
a.store.saveMessage(context.Background(), sessionID, "assistant", body)
} }
return // do not call persist() again — already persisted above
} }
persist() // final state — same row, updated one last time with the concluding text persist() // final state — same row, updated one last time with the concluding text
} }

View File

@@ -164,11 +164,29 @@ func handleChat(w http.ResponseWriter, r *http.Request, a *agent, st *store) {
http.Error(w, "bad request: "+err.Error(), 400) http.Error(w, "bad request: "+err.Error(), 400)
return return
} }
if req.Message == "" { if req.Message == "" && req.SessionID == "" {
http.Error(w, "message is required", 400) http.Error(w, "message is required", 400)
return return
} }
// Empty message with an existing session = reconnect/resume. The
// frontend sends this after a dropped SSE stream to re-establish the
// connection and catch up on any auto-continuation work that happened
// while disconnected. Route into resumeSession so the agent sees a
// system note and reports current state.
if req.Message == "" && req.SessionID != "" {
slog.Info("nomos: reconnect", "session", req.SessionID)
safego.Go("nomos:reconnect:"+req.SessionID, func() {
note := "[System: the operator's connection was re-established. The task may have progressed in the background — report your current state and progress.]"
a.resumeSession(context.Background(), req.SessionID, note)
})
// Return 202 so the frontend doesn't try to consume an SSE stream
// from this POST — resumeSession writes to the DB directly and
// the poller (already running from handleDisconnect) picks it up.
w.WriteHeader(202)
return
}
flusher, ok := w.(http.Flusher) flusher, ok := w.(http.Flusher)
if !ok { if !ok {
http.Error(w, "streaming not supported", 500) http.Error(w, "streaming not supported", 500)
@@ -318,6 +336,14 @@ func handleSessionDetail(w http.ResponseWriter, r *http.Request, st *store, a *a
return return
} }
// POST /sessions/{id}/resume — the operator asks the agent to continue.
if len(parts) == 2 && parts[1] == "resume" && r.Method == http.MethodPost {
note := "[System: the operator wants you to continue. Pick up where you left off — execute the next step of the plan, diagnose and fix any failures, or report progress if everything is done.]"
safego.Go("nomos:resume-session", func() { a.resumeSession(context.Background(), id, note) })
w.WriteHeader(202)
return
}
// GET /sessions/{id}/plan and /sessions/{id}/questions — REST hydration for // GET /sessions/{id}/plan and /sessions/{id}/questions — REST hydration for
// the context panel when it first opens a task; live events carry deltas // the context panel when it first opens a task; live events carry deltas
// from there. // from there.

View File

@@ -54,6 +54,7 @@ type session struct {
Outcome string `json:"outcome,omitempty"` Outcome string `json:"outcome,omitempty"`
Summary string `json:"summary,omitempty"` Summary string `json:"summary,omitempty"`
EntityID string `json:"entity_id,omitempty"` EntityID string `json:"entity_id,omitempty"`
PendingApprovals int `json:"pending_approvals"`
CreatedAt time.Time `json:"created_at"` CreatedAt time.Time `json:"created_at"`
LastActiveAt time.Time `json:"last_active_at"` LastActiveAt time.Time `json:"last_active_at"`
} }
@@ -195,9 +196,19 @@ func (s *store) listSessions(ctx context.Context) ([]session, error) {
return nil, nil return nil, nil
} }
rows, err := s.pool.Query(ctx, rows, err := s.pool.Query(ctx,
`SELECT id, title, actor, goal, status, COALESCE(outcome, ''), summary, `SELECT s.id, s.title, s.actor, s.goal, s.status, COALESCE(s.outcome, ''), s.summary,
COALESCE(entity_id::text, ''), created_at, last_active_at COALESCE(s.entity_id::text, ''),
FROM agent_sessions ORDER BY last_active_at DESC LIMIT 50`) COALESCE(pa.cnt, 0),
s.created_at, s.last_active_at
FROM agent_sessions s
LEFT JOIN (
SELECT l.session_id, COUNT(*) AS cnt
FROM nomos_plan_executions l
JOIN executions e ON e.entity_id = l.execution_id
WHERE e.status = 'pending_approval'
GROUP BY l.session_id
) pa ON pa.session_id = s.id
ORDER BY s.last_active_at DESC LIMIT 50`)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -207,7 +218,8 @@ func (s *store) listSessions(ctx context.Context) ([]session, error) {
for rows.Next() { for rows.Next() {
var sess session var sess session
if err := rows.Scan(&sess.ID, &sess.Title, &sess.Actor, &sess.Goal, &sess.Status, if err := rows.Scan(&sess.ID, &sess.Title, &sess.Actor, &sess.Goal, &sess.Status,
&sess.Outcome, &sess.Summary, &sess.EntityID, &sess.CreatedAt, &sess.LastActiveAt); err != nil { &sess.Outcome, &sess.Summary, &sess.EntityID, &sess.PendingApprovals,
&sess.CreatedAt, &sess.LastActiveAt); err != nil {
return nil, err return nil, err
} }
out = append(out, sess) out = append(out, sess)
@@ -215,6 +227,24 @@ func (s *store) listSessions(ctx context.Context) ([]session, error) {
return out, rows.Err() return out, rows.Err()
} }
func (s *store) getSession(ctx context.Context, id string) (*session, error) {
if s == nil {
return nil, nil
}
var sess session
err := s.pool.QueryRow(ctx,
`SELECT id, title, actor, goal, status, COALESCE(outcome, ''), summary,
COALESCE(entity_id::text, ''), 0, created_at, last_active_at
FROM agent_sessions WHERE id = $1`, id).
Scan(&sess.ID, &sess.Title, &sess.Actor, &sess.Goal, &sess.Status,
&sess.Outcome, &sess.Summary, &sess.EntityID, &sess.PendingApprovals,
&sess.CreatedAt, &sess.LastActiveAt)
if err != nil {
return nil, err
}
return &sess, nil
}
// getMessages returns a session's ENTIRE message history, unbounded — used // getMessages returns a session's ENTIRE message history, unbounded — used
// for the UI's own transcript view (GET /sessions/{id}), where the operator // for the UI's own transcript view (GET /sessions/{id}), where the operator
// should be able to see everything a task has done regardless of how long // should be able to see everything a task has done regardless of how long
@@ -387,6 +417,17 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
return nil, err return nil, err
} }
startSeq = 0 startSeq = 0
} else {
// Mid-flight plan revision: mark any still-pending steps from the
// previous generation as 'replaced' so the panel doesn't show them
// as incomplete forever. Only pending steps — already-running or
// done steps from the prior plan are preserved as history.
if _, err := tx.Exec(ctx, `
UPDATE session_plan_steps
SET status = 'replaced', finished_at = now()
WHERE session_id = $1 AND status = 'pending'`, sessionID); err != nil {
return nil, err
}
} }
out := make([]map[string]any, 0, len(steps)) out := make([]map[string]any, 0, len(steps))
@@ -428,6 +469,11 @@ func (s *store) proposePlan(ctx context.Context, sessionID string, steps []planS
// plan.step.finished (terminal) so the panel advances live. The execution link // 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 // is also what lets the api auto-close the step when the execution finishes
// (see closePlanStepForExecution). // (see closePlanStepForExecution).
//
// Completion ordering (done/failed/skipped/blocked) is enforced: a step cannot
// 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 string) error { func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, status, execID string) error {
if s == nil || sessionID == "" || sessionID == "ephemeral" { if s == nil || sessionID == "" || sessionID == "ephemeral" {
return nil return nil
@@ -436,9 +482,22 @@ func (s *store) updatePlanStep(ctx context.Context, sessionID string, seq int, s
switch status { switch status {
case "running": case "running":
stamp = ", started_at = COALESCE(started_at, now())" stamp = ", started_at = COALESCE(started_at, now())"
case "done", "failed", "skipped", "blocked": case "done", "failed", "skipped", "blocked", "replaced":
stamp = ", finished_at = now()" stamp = ", finished_at = now()"
} }
// Completion ordering: for terminal states, check that no earlier step
// is still pending. Running steps can start out of order (the agent
// may dispatch parallel work), but completion must be sequential.
if status == "done" || status == "failed" || status == "skipped" || status == "blocked" {
var blockedBy int
if err := s.pool.QueryRow(ctx, `
SELECT COALESCE(MIN(seq), 0)
FROM session_plan_steps
WHERE session_id = $1 AND seq < $2 AND status = 'pending'`,
sessionID, seq).Scan(&blockedBy); err == nil && blockedBy > 0 {
return fmt.Errorf("cannot complete step %d — step %d is still pending", seq, blockedBy)
}
}
var execPtr *uuid.UUID var execPtr *uuid.UUID
if id, err := uuid.Parse(execID); err == nil { if id, err := uuid.Parse(execID); err == nil {
execPtr = &id execPtr = &id
@@ -481,6 +540,33 @@ func (s *store) completeTask(ctx context.Context, sessionID, outcome, summary st
if s == nil || sessionID == "" || sessionID == "ephemeral" { if s == nil || sessionID == "" || sessionID == "ephemeral" {
return nil return nil
} }
// Auto-cancel any executions still in pending_approval/approved/queued
// state for this session — preventing orphaned approvals (observed in
// production: 4 approvals left open after session completed).
var cancelledCount int
if err := s.pool.QueryRow(ctx, `
WITH cancelled AS (
UPDATE executions SET status = 'cancelled',
result = '{"message": "task completed — auto-cancelled"}'::jsonb
WHERE entity_id IN (
SELECT execution_id FROM nomos_plan_executions WHERE session_id = $1
) AND status IN ('pending_approval', 'approved', 'queued')
RETURNING entity_id
)
SELECT COUNT(*) FROM cancelled
`, sessionID).Scan(&cancelledCount); err != nil {
slog.Warn("nomos: completeTask failed to cancel orphaned executions", "session", sessionID, "error", err)
}
// Mark all continuations done so the worker won't try to feed them back.
s.pool.Exec(ctx, `UPDATE nomos_plan_executions SET continued_at = now()
WHERE session_id = $1 AND continued_at IS NULL`, sessionID)
// Clean up assent and destructive window keys from autonomy_settings.
s.pool.Exec(ctx, `DELETE FROM autonomy_settings
WHERE key LIKE '%:' || $1`, sessionID)
status := "done" status := "done"
if outcome == "failure" { if outcome == "failure" {
status = "failed" status = "failed"
@@ -504,10 +590,27 @@ func (s *store) completeTask(ctx context.Context, sessionID, outcome, summary st
severity = "warning" severity = "warning"
} }
_ = observability.Event(ctx, sqlcgen.New(s.pool), "task.status", entPtr, severity, "nomos", sessionID, _ = observability.Event(ctx, sqlcgen.New(s.pool), "task.status", entPtr, severity, "nomos", sessionID,
map[string]any{"status": status, "outcome": outcome, "summary": summary}) map[string]any{"status": status, "outcome": outcome, "summary": summary,
"cancelled_executions": cancelledCount})
return nil return nil
} }
// 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 {
if s == nil || sessionID == "" {
return true // fail safe: don't warn when we can't check
}
var count int
s.pool.QueryRow(ctx, `
SELECT COUNT(*) FROM agent_activity
WHERE session_id = $1
AND tool_name IN ('update_entity_attributes', 'create_relationship')
AND success = true`, sessionID).Scan(&count)
return count > 0
}
// staleGoalSession is a goal-bearing task that's gone idle without reaching // staleGoalSession is a goal-bearing task that's gone idle without reaching
// a terminal state — the idle-sweep worker's work list (fix 2+3 of // a terminal state — the idle-sweep worker's work list (fix 2+3 of
// plans/2026-07-11-task-completion-safety-net.md). // plans/2026-07-11-task-completion-safety-net.md).

View File

@@ -202,7 +202,16 @@ func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args
if err != nil { if err != nil {
return fmt.Sprintf("error proposing plan: %v", err), true 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 // Nudge: if the last step doesn't mention entity writeback tools,
// the graph will keep drifting — discovered facts won't be persisted.
lastStep := steps[len(steps)-1]
hasWriteback := strings.Contains(lastStep.Title+lastStep.Detail, "update_entity_attributes") ||
strings.Contains(lastStep.Title+lastStep.Detail, "create_relationship")
result := fmt.Sprintf("Plan set: %d step(s). Execute them now, marking each with update_plan_step as you go.", len(persisted))
if !hasWriteback {
result += "\n\n⚠ The final step doesn't mention update_entity_attributes or create_relationship. Without those, any facts you discovered about entities (IPs, versions, hosts, states) will be LOST — the next session starts from scratch. Consider revising the last step to include entity writeback BEFORE completing the task."
}
return result, true
case "update_plan_step": case "update_plan_step":
seq := toInt(args["seq"]) seq := toInt(args["seq"])
@@ -261,7 +270,11 @@ func (a *agent) handleTaskTool(ctx context.Context, sessionID, name string, args
if err := a.store.completeTask(ctx, sessionID, outcome, summary); err != nil { if err := a.store.completeTask(ctx, sessionID, outcome, summary); err != nil {
return fmt.Sprintf("error completing task: %v", err), true return fmt.Sprintf("error completing task: %v", err), true
} }
return fmt.Sprintf("Task marked %s: %s", outcome, summary), true result := fmt.Sprintf("Task marked %s: %s", outcome, summary)
if !a.store.hadEntityWriteback(ctx, sessionID) {
result += "\n\n⚠ No entity attributes or relationships were updated in this session. Call update_entity_attributes and create_relationship to persist what you learned about entities before the next session starts from scratch."
}
return result, true
default: default:
return nil, false return nil, false
} }

View File

@@ -24,7 +24,6 @@ import (
"github.com/dtoro/oikos/internal/db/sqlcgen" "github.com/dtoro/oikos/internal/db/sqlcgen"
"github.com/dtoro/oikos/internal/observability" "github.com/dtoro/oikos/internal/observability"
"github.com/dtoro/oikos/internal/policy" "github.com/dtoro/oikos/internal/policy"
"github.com/dtoro/oikos/internal/safego"
"github.com/google/jsonschema-go/jsonschema" "github.com/google/jsonschema-go/jsonschema"
"github.com/google/uuid" "github.com/google/uuid"
"github.com/modelcontextprotocol/go-sdk/mcp" "github.com/modelcontextprotocol/go-sdk/mcp"
@@ -212,7 +211,7 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server {
InputSchema: objSchema( InputSchema: objSchema(
prop{"title", "string", "Short, specific, searchable title (e.g. 'Dragonfly memlock rlimit in unprivileged LXCs', not 'notes')."}, prop{"title", "string", "Short, specific, searchable title (e.g. 'Dragonfly memlock rlimit in unprivileged LXCs', not 'notes')."},
prop{"content", "string", "The knowledge itself, in markdown. Be concrete: symptom, root cause, the exact fix/commands, and any caveats. Written for someone hitting this fresh."}, prop{"content", "string", "The knowledge itself, in markdown. Be concrete: symptom, root cause, the exact fix/commands, and any caveats. Written for someone hitting this fresh."},
prop{"about", "string", "Optional entity slug this knowledge concerns (e.g. lxc:typetype, host:strong) — links the note to that entity so get_entity_knowledge surfaces it."}, prop{"about", "string", "Optional entity slug(s) this knowledge concerns. Pass a single slug (e.g. 'lxc:nfs-export') or a JSON array of slugs (e.g. '[\"lxc:nfs-export\", \"lxc:gitea\"]') to link to multiple entities. get_entity_knowledge surfaces it for each."},
prop{"tags", "string", "Optional comma-separated tags (e.g. 'docker,networking,gotcha')."}, prop{"tags", "string", "Optional comma-separated tags (e.g. 'docker,networking,gotcha')."},
prop{"kind", "string", "One of: investigation (a finding/incident analysis — default), document (reference), runbook (a repeatable procedure)."}, prop{"kind", "string", "One of: investigation (a finding/incident analysis — default), document (reference), runbook (a repeatable procedure)."},
), ),
@@ -358,186 +357,11 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server {
nStr(args["status"])), nil nStr(args["status"])), nil
}) })
register(&mcp.Tool{Name: "request_execution", Description: "Request a gated execution (agent-only mutation path). Supported actions: restart, systemctl, pct_exec, apt_upgrade, pct_create. pct_create provisions a NEW LXC and installs its service in one approved step — you do NOT need follow-up pct_exec calls for package installs.", // ── request_execution (legacy fixed enum) retired 2026-07-14 ──
InputSchema: objSchema( // All mutations now route through `run`. The handler functions
prop{"target", "string", "Target entity slug. For pct_create this MUST be the Proxmox HOST that will run the container (e.g. host:strong) — NOT the new LXC's name. For restart/systemctl/apt_upgrade/pct_exec use the target service/LXC slug (e.g. lxc:caddy)."}, // (runRexecRestart, runRexecSystemctl, etc.) are kept as reference
prop{"action", "string", "Action: restart, systemctl, pct_exec, apt_upgrade, pct_create"}, // for future runbook extraction — especially pct_create DNS/VMID logic.
prop{"params", "string", "For systemctl: 'enable|disable|reload'. For pct_exec: the shell command. For apt_upgrade: 'audit|upgrade'. For pct_create: a JSON object string with keys: vmid (int, required, unused id), hostname (string, required), cores (int), memory (MB int), disk_gb (int), ip (CIDR e.g. 192.168.8.50/24, or omit/\"dhcp\" — DHCP is the safe default, see below), gw (gateway ip, static only), bridge (e.g. vmbr0/vmbr1 — WHICH BRIDGE REACHES WHICH SUBNET IS DIFFERENT PER HOST, never assume vmbr0; see below), storage (default local-lvm), template (optional — omit to auto-pick newest debian on the host), privileged (bool), nesting (bool), mounts ([]string of 'src,mp=/dst'). Example: {\"vmid\":150,\"hostname\":\"typetype\",\"cores\":2,\"memory\":2048,\"disk_gb\":16,\"ip\":\"192.168.8.50/24\",\"gw\":\"192.168.8.2\",\"bridge\":\"vmbr1\",\"nesting\":true}. pct_create is ATOMIC — it ONLY creates and starts the container (no services/post_install params anymore). Once it completes you will be automatically re-invoked with the result; install packages and run setup by issuing your OWN `run` calls against the new lxc:<hostname> target, one step at a time — you'll see each step's real output and can fix exactly the one that fails, instead of one opaque multi-minute install that either fully works or fully doesn't. STATIC IP RULE: before setting ip/gw/bridge to anything other than DHCP, use list_entities/get_entity_knowledge to find an EXISTING lxc on the SAME host whose IP is in the same /28 block, and copy its exact gw+bridge — do not invent a gateway. If no such neighbor exists, prefer ip:\"dhcp\" (proven to work, gets a real routable address) over guessing; a wrong bridge/gateway pair fails a fast pre-flight ping check now (seconds, not minutes) but is still a wasted turn — better to not guess at all."}, // DO NOT re-register this tool. See plans/2026-07-10-general-gated-execution.md.
),
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
args := argsMap(req)
targetSlug, _ := args["target"].(string)
action, _ := args["action"].(string)
params, _ := args["params"].(string)
sessionID, _ := args["_session_id"].(string)
if targetSlug == "" || action == "" {
return textResult("error: target and action required"), nil
}
var targetID uuid.UUID
if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", targetSlug).Scan(&targetID); err != nil {
return textResult(fmt.Sprintf("target not found: %s", targetSlug)), nil
}
// restart, pct_exec, and systemctl (outside enable/disable) route
// through the same classify→gate path as `run` instead of executing
// immediately over SSH with a hardcoded risk_class='reversible_low'
// that was never actually checked against anything. Found live
// 2026-07-10: a chat request to "restart caddy" — the fleet's
// reverse proxy — executed instantly with zero approval, because
// this action bypassed the classifier entirely. classifyAndGate
// applies the same read-only/config-mutation/destructive
// classification and approval flow the `run` tool already uses.
if action == "restart" || action == "pct_exec" || (action == "systemctl" && params != "enable" && params != "disable") {
svc := strings.TrimPrefix(targetSlug, "lxc:")
var cmd, purpose string
switch action {
case "restart":
cmd = fmt.Sprintf("systemctl restart %s; sleep 1; systemctl is-active %s", svc, svc)
purpose = "restart " + svc
case "pct_exec":
cmd = params
purpose = "pct_exec (legacy) on " + targetSlug
case "systemctl":
cmd = fmt.Sprintf("systemctl %s %s; sleep 1; systemctl is-active %s", params, svc, svc)
purpose = "systemctl " + params + " " + svc
}
return classifyAndGate(ctx, pool, agentID, targetID, targetSlug, cmd, purpose, "", sessionID), nil
}
// Deduplicate: if a pending execution already exists for the same
// target+action, return the existing one instead of creating a
// duplicate. Prevents the LLM from re-requesting the same gated
// action in a tool-calling loop. Only blocks when a pending
// execution exists; completed/failed ones don't block.
if action == "systemctl" || action == "apt_upgrade" || action == "pct_create" {
execNamePrefix := action + " on " + targetSlug
var existingID string
err := pool.QueryRow(ctx, `
SELECT e.id::text FROM entities e
JOIN executions ex ON ex.entity_id = e.id
WHERE e.type = 'execution' AND e.name LIKE $1 AND ex.status = 'pending_approval'
ORDER BY e.created_at DESC LIMIT 1`, execNamePrefix+"%").Scan(&existingID)
if err == nil && existingID != "" {
return textResult(fmt.Sprintf("%s on %s is already queued for approval — execution %s. Wait for operator approval. Do not re-request.",
action, targetSlug, existingID)), nil
}
}
id, _ := uuid.NewV7()
correlationID := uuid.New().String()
// Full UUID, not a truncated prefix: UUIDv7's leading bytes encode a
// millisecond timestamp, so an 8-char prefix collides for real under
// back-to-back requests (observed live: two `run` calls seconds
// apart hit entities_slug_key). The full string is guaranteed unique.
execName := action + " on " + targetSlug + " (" + id.String() + ")"
execSlug := "exec:" + targetSlug + ":" + id.String()
_, err := pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`,
id, execSlug, execName)
if err != nil {
return textResult(fmt.Sprintf("error: failed to create execution: %v", err)), nil
}
pool.Exec(ctx, `INSERT INTO executions (entity_id, target_entity_id, action, risk_class, status, correlation_id, agent_id) VALUES ($1, $2, $3, 'reversible_low', 'running', $4, $5) ON CONFLICT DO NOTHING`,
id, targetID, action+":"+params, correlationID, agentID)
pool.Exec(ctx, `
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
SELECT $1, $2, 'targets', '{"by":"nomos"}'::jsonb, now()
WHERE NOT EXISTS (
SELECT 1 FROM relationships
WHERE source_id = $1 AND target_id = $2 AND type = 'targets' AND valid_to IS NULL)`,
id, targetID)
if sessionID != "" {
pool.Exec(ctx, `
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
SELECT t.id, $1, 'involves', '{"by":"nomos"}'::jsonb, now()
FROM entities t WHERE t.slug = $2
AND NOT EXISTS (
SELECT 1 FROM relationships
WHERE source_id = t.id AND target_id = $1 AND type = 'involves' AND valid_to IS NULL)`,
id, "task:"+sessionID)
}
// Execute reversible actions immediately. restart/pct_exec/systemctl
// (outside enable/disable) never reach here — they're routed through
// classifyAndGate above, before this dedup+insert block.
switch action {
case "systemctl":
// Only enable/disable reach this case now.
svc := strings.TrimPrefix(targetSlug, "lxc:")
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
createApproval(ctx, pool, id, targetID, "systemctl", svc+":"+params, "config_mutation")
return textResult(fmt.Sprintf("systemctl %s on %s requires approval — execution %s queued", params, svc, id)), nil
case "apt_upgrade":
if params == "audit" {
host, user, err := resolveHost(ctx, pool, targetSlug)
if err != nil {
return textResult(fmt.Sprintf("resolve: %v", err)), nil
}
out, err := sshExec(ctx, host, user, "apt update -qq 2>&1 >/dev/null; apt list --upgradable 2>/dev/null | tail -n +2 | wc -l; apt list --upgradable 2>/dev/null | tail -n +2 | head -20")
if err != nil {
return textResult(fmt.Sprintf("apt audit error: %v", err)), nil
}
return textResult("apt audit:\n" + out), nil
}
// During an active assent window, auto-approve.
if assentWindowActive(ctx, pool, agentID, sessionID) {
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
createApproval(ctx, pool, id, targetID, "apt_upgrade", params, "config_mutation")
// Do NOT pre-flip approvals/executions status here (that was
// the previous, broken "autoApprove" helper). DecideApproval
// (invoked below) is the ONE place that transitions
// pending_approval -> approved and dispatches the real SSH
// work — it specifically looks for status='pending_approval'
// to find what to run. Pre-flipping the status past that
// state meant DecideApproval's own lookup found nothing,
// silently no-opped, and the execution sat at 'approved'
// forever with nothing actually running. Found live: every
// assent-window auto-approved pct_create/apt_upgrade has
// never actually executed, via this exact bug. Calling
// executeApprovedViaAPI directly against the untouched
// pending_approval row makes this identical to the manual
// Approve-button path, just without a human click.
//
// context.Background(), NOT ctx: ctx is scoped to this MCP
// tool call, cancelled the instant the chat turn's HTTP
// response completes (every normal turn) — a goroutine
// meant to outlive the request must not inherit its context.
safego.Go("mcp:executeApprovedViaAPI:apt_upgrade", func() {
executeApprovedViaAPI(context.Background(), id, targetSlug, "apt_upgrade:"+params)
})
slog.Info("mcp: apt_upgrade auto-approved via assent window", "execution_id", id)
return textResult(fmt.Sprintf("apt_upgrade on %s auto-approved via assent window — execution %s running.", targetSlug, id)), nil
}
// upgrade requires approval — queue
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
createApproval(ctx, pool, id, targetID, "apt_upgrade", params, "config_mutation")
return textResult(fmt.Sprintf("apt_upgrade on %s requires approval — execution %s queued", targetSlug, id)), nil
case "pct_create":
// During an active assent window, auto-approve and execute
// instead of queuing — the operator already approved the plan.
if assentWindowActive(ctx, pool, agentID, sessionID) {
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
createApproval(ctx, pool, id, targetID, "pct_create", params, "config_mutation")
// See the apt_upgrade case above for why there's no
// pre-flip-status "autoApprove" step here anymore, and why
// this uses context.Background().
safego.Go("mcp:executeApprovedViaAPI:pct_create", func() {
executeApprovedViaAPI(context.Background(), id, targetSlug, "pct_create:"+params)
})
slog.Info("mcp: pct_create auto-approved via assent window", "execution_id", id)
return textResult(fmt.Sprintf("pct_create on %s auto-approved via assent window — execution %s running. The LXC is being provisioned now.", targetSlug, id)), nil
}
pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
createApproval(ctx, pool, id, targetID, "pct_create", params, "config_mutation")
return textResult(fmt.Sprintf("pct_create on %s requires approval — execution %s queued. The LXC will be provisioned once approved.", targetSlug, id)), nil
default:
return textResult(fmt.Sprintf("unknown action: %s. Supported: restart, systemctl, pct_exec, apt_upgrade, pct_create", action)), nil
}
})
register(&mcp.Tool{Name: "run", Description: "Run ANY shell command against any host or LXC. This is the general execution primitive — prefer it over asking the operator to run something manually, and don't wait for a matching fixed action to exist. Every command is automatically risk-classified: read-only inspection (cat, systemctl status, docker ps, journalctl, df, git status, ...) runs immediately; anything that changes state requires operator approval (granted by the operator replying \"go ahead\"/\"yes\" in chat, or via the Approve button); commands matching a destructive pattern (rm -rf, dd, mkfs, pct/qm destroy, DROP TABLE, reboot, piping curl into a shell, ...) always require approval regardless of what you declare. You cannot talk your way past the destructive check by declaring a lower risk.", register(&mcp.Tool{Name: "run", Description: "Run ANY shell command against any host or LXC. This is the general execution primitive — prefer it over asking the operator to run something manually, and don't wait for a matching fixed action to exist. Every command is automatically risk-classified: read-only inspection (cat, systemctl status, docker ps, journalctl, df, git status, ...) runs immediately; anything that changes state requires operator approval (granted by the operator replying \"go ahead\"/\"yes\" in chat, or via the Approve button); commands matching a destructive pattern (rm -rf, dd, mkfs, pct/qm destroy, DROP TABLE, reboot, piping curl into a shell, ...) always require approval regardless of what you declare. You cannot talk your way past the destructive check by declaring a lower risk.",
InputSchema: objSchema( InputSchema: objSchema(
@@ -661,17 +485,27 @@ func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server {
// ─── Phase 5: operational MCP tools ────────────────────────────── // ─── Phase 5: operational MCP tools ──────────────────────────────
register(&mcp.Tool{Name: "list_lxcs", Description: "List all LXC containers with ID, host, IP, and state", register(&mcp.Tool{Name: "list_lxcs", Description: "List all LXC containers with ID, host, IP, and state. Pass state=\"active\" to exclude destroyed/deprecated containers.",
InputSchema: objSchema(), InputSchema: objSchema(
prop{"state", "string", "Optional: filter by entity state (active, destroyed, …)"},
),
}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) { }, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
state, _ := argsMap(req)["state"].(string)
var statePtr *string
if state != "" {
statePtr = &state
}
return annotateJSONResult(queryRows(ctx, pool, ` return annotateJSONResult(queryRows(ctx, pool, `
SELECT e.slug, e.name, e.attributes->>'pve_id' AS pve_id, SELECT e.slug, e.name, e.attributes->>'pve_id' AS pve_id,
e.attributes->>'lan_ip' AS lan_ip, e.attributes->>'lan_ip' AS lan_ip,
e.state,
st.health, st.last_check_at st.health, st.last_check_at
FROM entities e FROM entities e
LEFT JOIN entity_status st ON st.entity_id = e.id LEFT JOIN entity_status st ON st.entity_id = e.id
WHERE e.type = 'lxc' WHERE e.type = 'lxc'
ORDER BY (e.attributes->>'pve_id')::int`), "lxc_list"), nil AND ($1::text IS NULL OR e.state = $1)
ORDER BY CASE WHEN e.state = 'active' THEN 0 ELSE 1 END,
(e.attributes->>'pve_id')::int`, statePtr), "lxc_list"), nil
}) })
register(&mcp.Tool{Name: "ping_service", Description: "Check if a service is reachable via HTTP", register(&mcp.Tool{Name: "ping_service", Description: "Check if a service is reachable via HTTP",
@@ -1657,10 +1491,26 @@ func knowledgeSlug(kind, title string) string {
func upsertKnowledge(ctx context.Context, pool *db.Pool, args map[string]any) (*mcp.CallToolResult, error) { func upsertKnowledge(ctx context.Context, pool *db.Pool, args map[string]any) (*mcp.CallToolResult, error) {
title, _ := args["title"].(string) title, _ := args["title"].(string)
content, _ := args["content"].(string) content, _ := args["content"].(string)
about, _ := args["about"].(string)
tagsRaw, _ := args["tags"].(string) tagsRaw, _ := args["tags"].(string)
kind, _ := args["kind"].(string) kind, _ := args["kind"].(string)
// Normalize about: accept a single string slug or an array of slugs.
var aboutSlugs []string
switch v := args["about"].(type) {
case string:
if s := strings.TrimSpace(v); s != "" {
aboutSlugs = []string{s}
}
case []interface{}:
for _, item := range v {
if s, ok := item.(string); ok {
if s = strings.TrimSpace(s); s != "" {
aboutSlugs = append(aboutSlugs, s)
}
}
}
}
title = strings.TrimSpace(title) title = strings.TrimSpace(title)
content = strings.TrimSpace(content) content = strings.TrimSpace(content)
if title == "" || content == "" { if title == "" || content == "" {
@@ -1707,11 +1557,13 @@ func upsertKnowledge(ctx context.Context, pool *db.Pool, args map[string]any) (*
return textResult(fmt.Sprintf("error writing knowledge: %v", err)), nil return textResult(fmt.Sprintf("error writing knowledge: %v", err)), nil
} }
// Link it to the entity it's about, if given and not already linked. // Link it to the entity(s) it's about, if given and not already linked.
linked := "" linked := ""
if about = strings.TrimSpace(about); about != "" { if len(aboutSlugs) > 0 {
var linkedSlugs []string
for _, slug := range aboutSlugs {
var targetID uuid.UUID var targetID uuid.UUID
if qerr := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", about).Scan(&targetID); qerr == nil { if qerr := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", slug).Scan(&targetID); qerr == nil {
pool.Exec(ctx, ` pool.Exec(ctx, `
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from) INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
SELECT $1, $2, 'documents', '{"by":"nomos"}'::jsonb, now() SELECT $1, $2, 'documents', '{"by":"nomos"}'::jsonb, now()
@@ -1719,9 +1571,13 @@ func upsertKnowledge(ctx context.Context, pool *db.Pool, args map[string]any) (*
SELECT 1 FROM relationships SELECT 1 FROM relationships
WHERE source_id = $1 AND target_id = $2 AND type = 'documents' AND valid_to IS NULL)`, WHERE source_id = $1 AND target_id = $2 AND type = 'documents' AND valid_to IS NULL)`,
docID, targetID) docID, targetID)
linked = " and linked to " + about linkedSlugs = append(linkedSlugs, slug)
} else { }
linked = fmt.Sprintf(" (note: entity %q not found, saved unlinked)", about) }
if len(linkedSlugs) == 1 {
linked = " and linked to " + linkedSlugs[0]
} else if len(linkedSlugs) > 1 {
linked = fmt.Sprintf(" and linked to %d entities", len(linkedSlugs))
} }
} }

View File

@@ -0,0 +1,17 @@
-- 020_session_reliability.up.sql
-- Plan step generation tracking + audit log session linkage.
-- See plans/2026-07-14-session-reliability-and-ux-audit.md.
-- Plan step generation: when the agent revises a plan mid-flight, new steps
-- get a higher generation number so the frontend can group/collapse old ones.
ALTER TABLE session_plan_steps ADD COLUMN IF NOT EXISTS generation INTEGER NOT NULL DEFAULT 1;
-- Track which session produced each audit-log entry so per-session analysis
-- (e.g. "did this task call update_entity_attributes?") is O(1) instead of
-- scanning the full log.
ALTER TABLE audit_log ADD COLUMN IF NOT EXISTS session_id UUID;
CREATE INDEX IF NOT EXISTS idx_audit_log_session ON audit_log (session_id);
-- Efficient lookup of pending-approval executions by session.
CREATE INDEX IF NOT EXISTS idx_nomos_plan_executions_session
ON nomos_plan_executions (session_id) WHERE continued_at IS NULL;

View File

@@ -48,6 +48,26 @@ of what a command does.
## Every chat is a task ## Every chat is a task
### ⚠️ AFTER EVERY TASK: WRITE BACK OR LOSE IT
What you discovered but didn't write back is **lost** — the next session starts
from scratch. Before calling `complete_task`, you MUST:
1. `update_entity_attributes` — ANY concrete fact (IP, version, host, port,
state) for ANY entity you learned about. Every LXC you queried, every target
you ran against. Nothing in your transcript survives — only attributes do.
2. `create_relationship` — ANY edge you discovered (hosts, depends-on,
provides). Every "X runs on Y" fact.
3. `upsert_knowledge` — the narrative: what you did, what broke, the fix.
Link to ALL affected entities via `about` (pass an array).
**The plan's LAST step must list these by name.** Not "record findings" —
"1. update_entity_attributes for each audited LXC, 2. create_relationship
for any discovered host/container edges, 3. upsert_knowledge." Future you
depends on this.
---
Each conversation is a **task**: a goal the operator wants achieved, from Each conversation is a **task**: a goal the operator wants achieved, from
"install service X" to "give me the key status of Y". Every non-trivial task "install service X" to "give me the key status of Y". Every non-trivial task
has the SAME first step and the SAME last step — research in, knowledge out — has the SAME first step and the SAME last step — research in, knowledge out —
@@ -125,10 +145,10 @@ exist just to fill the step. The loop scales down; it doesn't disappear.
can be multi-line), `purpose` (one sentence — the operator sees exactly this when can be multi-line), `purpose` (one sentence — the operator sees exactly this when
deciding). Auto-runs if read-only; otherwise queues for approval. See "Your deciding). Auto-runs if read-only; otherwise queues for approval. See "Your
capability is unlimited" above. capability is unlimited" above.
- `request_execution` — curated fast-paths for common named actions: restart, systemctl - `run` — the ONLY mutation tool. Accepts `target`, `command`, `purpose`,
(enable/disable/reload), pct_exec (shell command inside an existing LXC), apt_upgrade `declared_risk`. The `request_execution` fixed-enum tool is RETIRED
(audit/upgrade), pct_create (provision a new LXC). Use these when they fit; use `run` (2026-07-14) — use `run` for EVERYTHING: restarts, apt upgrades, pct exec,
for everything else — you do not need a matching named action to act. pct create, any shell command. There is no named-action tool anymore.
- `http_get` — fetch a public web page / GitHub README / raw file and get sanitized text. - `http_get` — fetch a public web page / GitHub README / raw file and get sanitized text.
You CAN read the internet with this. When asked to deploy a service from a URL or repo, You CAN read the internet with this. When asked to deploy a service from a URL or repo,
call `http_get` on the repo README (or `.../raw/main/docker-compose.yml`) to learn its call `http_get` on the repo README (or `.../raw/main/docker-compose.yml`) to learn its
@@ -158,7 +178,7 @@ exist just to fill the step. The loop scales down; it doesn't disappear.
## Policy awareness ## Policy awareness
Before calling `request_execution`: Before calling `run`:
- Check risk class via `get_entity` on the target - Check risk class via `get_entity` on the target
- `pct_create``config_mutation`: **ATOMIC** — creates and starts a new LXC, nothing - `pct_create``config_mutation`: **ATOMIC** — creates and starts a new LXC, nothing
more. Set `target` to the Proxmox HOST slug (e.g. `host:strong`), not the new container more. Set `target` to the Proxmox HOST slug (e.g. `host:strong`), not the new container
@@ -233,7 +253,7 @@ note — continue executing the full plan from there. Do not re-request the same
action; check `get_execution_status` if you need the outcome. One approval per action; check `get_execution_status` if you need the outcome. One approval per
action is enough. action is enough.
**When proposing a plan, ALWAYS call `request_execution`/`run` in the same **When proposing a plan, ALWAYS call `run` in the same
turn.** Do not propose a plan in text, ask "shall I proceed?", and wait. turn.** Do not propose a plan in text, ask "shall I proceed?", and wait.
Call the tool — if it queues for approval, present what's queued and stop. Call the tool — if it queues for approval, present what's queued and stop.
The operator's "proceed"/"go ahead" will grant it and open the assent window. The operator's "proceed"/"go ahead" will grant it and open the assent window.

View File

@@ -7,8 +7,9 @@
## Overview ## Overview
Standard operating procedures for the Nomos agent managing the hubris Standard operating procedures for the Nomos agent managing the hubris
homelab. All mutations route through `request_execution` → Oikos policy homelab. All mutations route through `run` → Oikos policy
gating → actuator (SSH). gating → actuator (SSH). The `request_execution` fixed-enum tool was retired
2026-07-14.
## Procedures ## Procedures
@@ -22,13 +23,13 @@ gating → actuator (SSH).
### Signal response ### Signal response
- `reversible_low` with validated pattern → `request_execution` (auto-restart) - `reversible_low` with validated pattern → `run` (auto-restart)
- `config_mutation` or `destructive` → escalate to operator - `config_mutation` or `destructive` → escalate to operator
- Repeated flapping → escalate with flap count - Repeated flapping → escalate with flap count
### Execution tracking ### Execution tracking
1. `request_execution` returns a correlation_id 1. `run` returns the execution ID in its result text
2. Poll `get_event_timeline` filtering by correlation_id 2. Poll `get_event_timeline` filtering by correlation_id
3. Once complete, `get_health_summary` to verify recovery 3. Once complete, `get_health_summary` to verify recovery
4. Record outcome via internal reasoning 4. Record outcome via internal reasoning
@@ -41,7 +42,9 @@ gating → actuator (SSH).
## Changelog ## Changelog
### 2026-07-08 — rename to Nomos ### 2026-07-14 — request_execution retired
All references to `request_execution` replaced with `run`. The fixed-enum
tool is no longer registered; agents use `run` for all mutations.
Agent renamed from Hermes to Nomos (N0 milestone). Agent renamed from Hermes to Nomos (N0 milestone).
### 2026-07-07 — initial Phase 4 skill ### 2026-07-07 — initial Phase 4 skill

File diff suppressed because it is too large Load Diff

View File

@@ -14,6 +14,7 @@ went sideways, open an investigation.
| 2026-07-08 | [Liveness, drift, and UX cohesion](2026-07-08-liveness-drift-and-ux-cohesion.md) | In Progress — Phase 5 deferred | | 2026-07-08 | [Liveness, drift, and UX cohesion](2026-07-08-liveness-drift-and-ux-cohesion.md) | In Progress — Phase 5 deferred |
| 2026-07-10 | [General gated execution: unlimited actions, gated by risk](2026-07-10-general-gated-execution.md) | In Progress — enum retirement + auto-act revival still open | | 2026-07-10 | [General gated execution: unlimited actions, gated by risk](2026-07-10-general-gated-execution.md) | In Progress — enum retirement + auto-act revival still open |
| 2026-07-11 | [Nomos agent code review: gaps and improvement plan](2026-07-11-nomos-agent-code-review.md) | In Progress — only C1 (unauthenticated nomos gateway) still open, deferred | | 2026-07-11 | [Nomos agent code review: gaps and improvement plan](2026-07-11-nomos-agent-code-review.md) | In Progress — only C1 (unauthenticated nomos gateway) still open, deferred |
| 2026-07-14 | [Session reliability & UX audit](2026-07-14-session-reliability-and-ux-audit.md) | Planned — just audited, not started |
## Done ## Done

View File

@@ -20,6 +20,7 @@ export interface Session {
outcome?: string // success | failure | partial outcome?: string // success | failure | partial
summary?: string summary?: string
entity_id?: string entity_id?: string
pending_approvals?: number
created_at: string created_at: string
last_active_at: string last_active_at: string
} }
@@ -51,12 +52,18 @@ export async function deleteSession(sessionId: string): Promise<boolean> {
return res.ok return res.ok
} }
export async function resumeSession(sessionId: string): Promise<boolean> {
const res = await fetchWithAuth(`${BASE}/sessions/${sessionId}/resume`, { method: 'POST' })
return res.ok
}
export interface PlanStep { export interface PlanStep {
id: string id: string
seq: number seq: number
title: string title: string
detail: string detail: string
status: 'pending' | 'running' | 'done' | 'failed' | 'skipped' | 'blocked' status: 'pending' | 'running' | 'done' | 'failed' | 'skipped' | 'blocked' | 'replaced'
generation?: number
execution_id?: string execution_id?: string
target_slug?: string target_slug?: string
started_at?: string started_at?: string

View File

@@ -7,19 +7,32 @@
import CircleXIcon from '@lucide/svelte/icons/circle-x' import CircleXIcon from '@lucide/svelte/icons/circle-x'
import CircleSlashIcon from '@lucide/svelte/icons/circle-slash' import CircleSlashIcon from '@lucide/svelte/icons/circle-slash'
import CirclePauseIcon from '@lucide/svelte/icons/circle-pause' import CirclePauseIcon from '@lucide/svelte/icons/circle-pause'
import ChevronDownIcon from '@lucide/svelte/icons/chevron-down'
import ChevronRightIcon from '@lucide/svelte/icons/chevron-right'
const done = $derived($planSteps.filter((s) => s.status === 'done').length) const done = $derived($planSteps.filter((s) => s.status === 'done').length)
const total = $derived($planSteps.length) const total = $derived($planSteps.length)
const pct = $derived(total > 0 ? Math.round((done / total) * 100) : 0) const pct = $derived(total > 0 ? Math.round((done / total) * 100) : 0)
// Group steps by generation. The latest generation is shown expanded;
// older ones are collapsible.
const byGeneration = $derived.by(() => {
const groups: Map<number, typeof $planSteps> = new Map()
for (const s of $planSteps) {
const g = s.generation ?? 1
if (!groups.has(g)) groups.set(g, [])
groups.get(g)!.push(s)
}
return Array.from(groups.entries()).sort(([a], [b]) => a - b)
})
const latestGen = $derived(byGeneration.length > 0 ? byGeneration[byGeneration.length - 1][0] : 0)
let openGens = $state(new Set<number>())
let sheetSlug = $state<string | null>(null) let sheetSlug = $state<string | null>(null)
let sheetOpen = $state(false) let sheetOpen = $state(false)
// Tool calls don't carry a step id, so a step can't be linked to its exact
// transcript entry — but its target entity IS known, and EntitySheet already
// gives a real, working detail view for any slug. Clicking a step with a
// target opens that, rather than a fake "scroll to it" that would silently
// no-op for a collapsed tool-call group.
function openStep(targetSlug: string | undefined) { function openStep(targetSlug: string | undefined) {
if (!targetSlug) return if (!targetSlug) return
sheetSlug = targetSlug sheetSlug = targetSlug
@@ -36,12 +49,30 @@
<div class="h-1 w-full overflow-hidden rounded-full bg-muted"> <div class="h-1 w-full overflow-hidden rounded-full bg-muted">
<div class="h-full rounded-full bg-primary transition-all duration-500" style="width: {pct}%"></div> <div class="h-full rounded-full bg-primary transition-all duration-500" style="width: {pct}%"></div>
</div> </div>
{#each byGeneration as [gen, steps] (gen)}
{@const isLatest = gen === latestGen}
{#if byGeneration.length > 1}
<button
type="button"
class="flex items-center gap-1 text-[10px] text-muted-foreground hover:text-foreground"
onclick={() => openGens.has(gen) ? openGens.delete(gen) : openGens.add(gen); openGens = new Set(openGens)}
>
{#if openGens.has(gen) || isLatest}
<ChevronDownIcon class="size-3" />
{:else}
<ChevronRightIcon class="size-3" />
{/if}
<span>{isLatest ? 'Current plan' : `Plan v${gen} (replaced)`}</span>
</button>
{/if}
{#if isLatest || openGens.has(gen)}
<ol class="flex flex-col gap-1"> <ol class="flex flex-col gap-1">
{#each $planSteps as step (step.id)} {#each steps as step (step.id)}
<li> <li>
<button <button
type="button" type="button"
class="flex w-full items-start gap-2 rounded px-1 py-1 text-left text-xs {step.target_slug ? 'hover:bg-muted/50' : 'cursor-default'}" class="flex w-full items-start gap-2 rounded px-1 py-1 text-left text-xs {step.target_slug ? 'hover:bg-muted/50' : 'cursor-default'} {gen !== latestGen ? 'opacity-50' : ''}"
onclick={() => openStep(step.target_slug)} onclick={() => openStep(step.target_slug)}
> >
<span class="mt-0.5 shrink-0"> <span class="mt-0.5 shrink-0">
@@ -51,7 +82,7 @@
<CircleXIcon class="size-3.5 text-destructive" /> <CircleXIcon class="size-3.5 text-destructive" />
{:else if step.status === 'running'} {:else if step.status === 'running'}
<LoaderCircleIcon class="size-3.5 animate-spin text-primary" /> <LoaderCircleIcon class="size-3.5 animate-spin text-primary" />
{:else if step.status === 'skipped'} {:else if step.status === 'skipped' || step.status === 'replaced'}
<CircleSlashIcon class="size-3.5 text-muted-foreground" /> <CircleSlashIcon class="size-3.5 text-muted-foreground" />
{:else if step.status === 'blocked'} {:else if step.status === 'blocked'}
<CirclePauseIcon class="size-3.5 text-warning" /> <CirclePauseIcon class="size-3.5 text-warning" />
@@ -71,6 +102,8 @@
</li> </li>
{/each} {/each}
</ol> </ol>
{/if}
{/each}
</div> </div>
{/if} {/if}

View File

@@ -16,18 +16,27 @@
// still refetch once outcome/summary land, not just on session switch. // still refetch once outcome/summary land, not just on session switch.
let loadedKey = $state<string | null>(null) let loadedKey = $state<string | null>(null)
// Reload the digest whenever the session changes, the task's status changes let pollTimer: ReturnType<typeof setInterval> | null = $state(null)
// (e.g. it just completed), or a stream finishes — "what did this session
// actually do" is only meaningful once executions have had a chance to land.
$effect(() => { $effect(() => {
const sid = $currentSession const sid = $currentSession
const busy = $streaming const busy = $streaming
const status = $currentTask?.status ?? '' const status = $currentTask?.status ?? ''
if (!sid || busy) return if (!sid || busy) return
const key = `${sid}:${status}` const key = `${sid}:${status}`
if (loadedKey === key) return if (loadedKey !== key) {
loadedKey = key loadedKey = key
fetchSessionDigest(sid).then((d) => (digest = d)) fetchSessionDigest(sid).then((d) => (digest = d))
}
// Poll every 10s while the session is active.
if (status !== 'done' && status !== 'failed') {
if (!pollTimer) pollTimer = setInterval(() => fetchSessionDigest(sid).then((d) => (digest = d)), 10000)
} else {
if (pollTimer) { clearInterval(pollTimer); pollTimer = null }
}
return () => {
if (pollTimer) { clearInterval(pollTimer); pollTimer = null }
}
}) })
function statusVariant(status: string): 'default' | 'secondary' | 'destructive' | 'outline' { function statusVariant(status: string): 'default' | 'secondary' | 'destructive' | 'outline' {

View File

@@ -51,7 +51,12 @@
onclick={() => handleClick(session.id)} onclick={() => handleClick(session.id)}
> >
<span class="min-w-0 max-w-full truncate font-medium">{session.title || 'Untitled'}</span> <span class="min-w-0 max-w-full truncate font-medium">{session.title || 'Untitled'}</span>
<span class="text-[11px] text-muted-foreground">{relativeTime(session.last_active_at)}</span> <span class="flex items-center gap-1.5 text-[11px] text-muted-foreground">
{relativeTime(session.last_active_at)}
{#if session.pending_approvals}
<span class="rounded bg-warning/20 px-1 text-[10px] font-medium text-warning">{session.pending_approvals}</span>
{/if}
</span>
</button> </button>
<button <button
type="button" type="button"

View File

@@ -0,0 +1,108 @@
<script lang="ts">
import type { ToolCallResult } from '$lib/stores/chat'
import { getExecution } from '$lib/api'
import LoaderCircleIcon from '@lucide/svelte/icons/loader-circle'
import CheckIcon from '@lucide/svelte/icons/check'
import XIcon from '@lucide/svelte/icons/x'
import ClockIcon from '@lucide/svelte/icons/clock'
let { tool }: { tool: ToolCallResult } = $props()
const loading = $derived(tool.type === 'tool_use')
const error = $derived(tool.type === 'tool_result' ? tool.error : undefined)
const executions = $derived.by(() => {
if (tool.type !== 'tool_result' || !tool.result) return null
const data = Array.isArray(tool.result) ? tool.result : (tool.result as any)?.data
return Array.isArray(data) ? data as any[] : null
})
const statusColors: Record<string, string> = {
completed: 'var(--success)',
failed: 'var(--destructive)',
cancelled: 'var(--destructive)',
denied: 'var(--destructive)',
revoked: 'var(--destructive)',
running: 'var(--warning)',
approved: 'var(--warning)',
pending_approval: 'var(--muted-foreground)',
queued: 'var(--muted-foreground)',
}
function statusLabel(s: string): string {
return s.replace(/_/g, ' ').replace(/\b\w/g, (c) => c.toUpperCase())
}
function fmtDuration(ms?: number): string {
if (!ms) return ''
const s = Math.round(ms / 1000)
if (s < 60) return `${s}s`
return `${Math.floor(s / 60)}m ${s % 60}s`
}
function truncate(s: string, n: number): string {
if (!s) return ''
return s.length > n ? s.slice(0, n) + '…' : s
}
</script>
{#if loading}
<div class="flex items-center gap-2 rounded-lg border bg-card px-3 py-2 text-xs" role="status">
<LoaderCircleIcon class="size-3 shrink-0 animate-spin text-primary" aria-hidden="true" />
<span class="font-medium">Execution status</span>
<span class="animate-pulse text-muted-foreground">checking…</span>
</div>
{:else if error}
<div class="flex items-center gap-2 rounded-lg border border-destructive/30 bg-destructive/5 px-3 py-2 text-xs" role="alert">
<XIcon class="size-3 shrink-0 text-destructive" aria-hidden="true" />
<span class="font-medium">Execution status</span>
<span class="text-destructive">{error}</span>
</div>
{:else if executions && executions.length > 0}
<div class="rounded-lg border bg-card text-xs">
<div class="flex flex-col divide-y">
{#each executions as exec (exec.execution_id ?? exec.id)}
{@const status = exec.status ?? 'unknown'}
{@const color = statusColors[status] ?? 'var(--muted-foreground)'}
{@const isRunning = status === 'running' || status === 'approved'}
<div class="flex items-center gap-2 px-3 py-2">
{#if status === 'completed'}
<CheckIcon class="size-3 shrink-0" style="color: {color}" aria-hidden="true" />
{:else if status === 'failed' || status === 'cancelled' || status === 'denied' || status === 'revoked'}
<XIcon class="size-3 shrink-0" style="color: {color}" aria-hidden="true" />
{:else if isRunning}
<LoaderCircleIcon class="size-3 shrink-0 animate-spin" style="color: {color}" aria-hidden="true" />
{:else}
<ClockIcon class="size-3 shrink-0 text-muted-foreground" aria-hidden="true" />
{/if}
<span class="font-mono font-medium">{truncate(exec.execution_id ?? exec.id ?? '', 12)}</span>
<span class="text-muted-foreground">{statusLabel(status)}</span>
{#if exec.action}
<span class="text-muted-foreground">· {truncate(exec.action, 40)}</span>
{/if}
{#if exec.duration_ms}
<span class="text-muted-foreground">· {fmtDuration(exec.duration_ms)}</span>
{/if}
<span class="ml-auto inline-block rounded px-1.5 py-0.5 font-medium text-[10px]" style="background: {color}22; color: {color}">
{statusLabel(status)}
</span>
</div>
{#if exec.result || exec.error}
<div class="max-h-32 overflow-y-auto bg-background/60 px-3 py-1.5">
{#if exec.error}
<pre class="whitespace-pre-wrap break-all font-mono text-[11px] text-destructive">{exec.error}</pre>
{:else if exec.result}
<pre class="whitespace-pre-wrap break-all font-mono text-[11px] text-muted-foreground">{typeof exec.result === 'string' ? exec.result : JSON.stringify(exec.result, null, 2)}</pre>
{/if}
</div>
{/if}
{/each}
</div>
</div>
{:else}
<div class="flex items-center gap-2 rounded-lg border bg-card px-3 py-2 text-xs" role="status">
<CheckIcon class="size-3 shrink-0 text-success" aria-hidden="true" />
<span class="font-medium">Execution status</span>
<span class="text-muted-foreground">no active executions</span>
</div>
{/if}

View File

@@ -0,0 +1,9 @@
import { registerToolRenderer } from '$lib/tool-renderers'
import ExecutionStatus from './ExecutionStatus.svelte'
export function init() {
registerToolRenderer({
match: (t) => t.name === 'get_execution_status',
component: ExecutionStatus,
})
}

View File

@@ -7,6 +7,7 @@ import { init as initBlastRadius } from './blast-radius'
import { init as initChangeLog } from './change-log' import { init as initChangeLog } from './change-log'
import { init as initFleetSnapshot } from './fleet-snapshot' import { init as initFleetSnapshot } from './fleet-snapshot'
import { init as initMetricChart } from './metric-chart' import { init as initMetricChart } from './metric-chart'
import { init as initExecutionStatus } from './execution-status'
initEntityCard() initEntityCard()
initHealthSummary() initHealthSummary()
@@ -17,3 +18,4 @@ initBlastRadius()
initChangeLog() initChangeLog()
initFleetSnapshot() initFleetSnapshot()
initMetricChart() initMetricChart()
initExecutionStatus()

View File

@@ -40,7 +40,7 @@ function extractApprovals(tools: ToolCallResult[]): PendingApproval[] {
if (m) { if (m) {
out.push({ out.push({
executionId: m[1], executionId: m[1],
action: t.args?.action ?? t.args?.purpose ?? t.name ?? 'unknown', action: t.args?.purpose?.slice(0, 60) ?? t.args?.action ?? t.name ?? 'unknown',
target: t.args?.target ?? 'unknown', target: t.args?.target ?? 'unknown',
destructive: /\bDESTRUCTIVE\b/.test(text), destructive: /\bDESTRUCTIVE\b/.test(text),
command: t.args?.command, command: t.args?.command,
@@ -66,10 +66,20 @@ function mid(): string {
export const messages = writable<ChatMessage[]>([]) export const messages = writable<ChatMessage[]>([])
export const streaming = writable(false) export const streaming = writable(false)
export const connectionState = writable<'connected' | 'disconnected' | 'reconnecting'>('connected')
export const currentSession = writable<string | null>(null) export const currentSession = writable<string | null>(null)
export const sessions = writable<Session[]>([]) export const sessions = writable<Session[]>([])
export const sessionMessages = writable<Message[]>([]) export const sessionMessages = writable<Message[]>([])
export const error = writable<string | null>(null) export const error = writable<string | null>(null)
export const chatErrors = writable<{ id: string; message: string; action?: string }[]>([])
export function dismissError(id: string) {
chatErrors.update((e) => e.filter((x) => x.id !== id))
}
export function addChatError(message: string, action?: string) {
chatErrors.update((e) => [...e, { id: crypto.randomUUID(), message, action }])
}
// Per-session controller tracking. Multiple tasks can stream concurrently // Per-session controller tracking. Multiple tasks can stream concurrently
// (see sendMessage's session guard above this used to be a single global // (see sendMessage's session guard above this used to be a single global
@@ -153,7 +163,9 @@ function startPolling(sessionId: string) {
stopPolling() stopPolling()
pollingSessionId = sessionId pollingSessionId = sessionId
pollTimer = setInterval(async () => { pollTimer = setInterval(async () => {
if (get(streaming)) return // Allow polling while disconnected — the agent is still working
// server-side and the poller is the only way to see it.
if (get(streaming) && get(connectionState) === 'connected') return
if (pollingSessionId !== sessionId || get(currentSession) !== sessionId) return if (pollingSessionId !== sessionId || get(currentSession) !== sessionId) return
const msgs = await fetchMessages(sessionId) const msgs = await fetchMessages(sessionId)
if (get(streaming) || pollingSessionId !== sessionId) return // re-check: the fetch itself takes time if (get(streaming) || pollingSessionId !== sessionId) return // re-check: the fetch itself takes time
@@ -217,6 +229,7 @@ export function sendMessage(text: string) {
// auto-continuation. // auto-continuation.
const openedFor = get(currentSession) const openedFor = get(currentSession)
let streamSessionID = openedFor let streamSessionID = openedFor
let receivedDone = false
const controller = streamChat( const controller = streamChat(
text, text,
@@ -271,6 +284,7 @@ export function sendMessage(text: string) {
last.tools = last.tools.map((t) => last.tools = last.tools.map((t) =>
t.id === ev.data.id ? updated : t t.id === ev.data.id ? updated : t
) )
last.pendingApprovals = extractApprovals(last.tools)
} }
return [...ms] return [...ms]
}) })
@@ -293,6 +307,8 @@ export function sendMessage(text: string) {
return [...ms] return [...ms]
}) })
} else if (ev.type === 'done') { } else if (ev.type === 'done') {
receivedDone = true
connectionState.set('connected')
messages.update((ms) => { messages.update((ms) => {
const last = ms[ms.length - 1] const last = ms[ms.length - 1]
if (last && last.role === 'assistant') { if (last && last.role === 'assistant') {
@@ -314,14 +330,29 @@ export function sendMessage(text: string) {
} }
}, },
(err: string) => { (err: string) => {
if (get(currentSession) === streamSessionID) error.set(err) // Distinguish user abort from network drop.
if (err === 'AbortError' || err.includes('aborted')) {
if (get(currentSession) === streamSessionID) streaming.set(false)
return
}
// Network blip / server restart — initiate reconnect.
if (get(currentSession) === streamSessionID) {
error.set(err)
if (!receivedDone && streamSessionID) {
handleDisconnect(streamSessionID)
} else {
streaming.set(false)
}
}
}, },
() => { () => {
if (get(currentSession) === streamSessionID) streaming.set(false) // SSE stream completed without error. If we never received 'done',
// Clean up whichever slot this controller ended up in — normally // the connection was severed mid-turn — treat as disconnect.
// activeControllers[streamSessionID] once the 'session' event has if (!receivedDone && streamSessionID && get(currentSession) === streamSessionID) {
// fired, but fall back to pendingController for the (rare) case where handleDisconnect(streamSessionID)
// the stream errored/completed before ever getting one. } else if (get(currentSession) === streamSessionID) {
streaming.set(false)
}
if (streamSessionID && activeControllers.get(streamSessionID) === controller) { if (streamSessionID && activeControllers.get(streamSessionID) === controller) {
activeControllers.delete(streamSessionID) activeControllers.delete(streamSessionID)
} }
@@ -340,12 +371,87 @@ export function sendMessage(text: string) {
} }
} }
// handleDisconnect is called when the SSE stream drops mid-turn without
// receiving a 'done' event. Falls back to polling and attempts reconnection.
function handleDisconnect(sessionId: string) {
const MAX_RECONNECT = 3
connectionState.set('disconnected')
startPolling(sessionId)
addChatError('Agent connection lost. The task is still running — retrying…', 'Dismiss')
let attempts = 0
let delay = 1000
const attemptReconnect = () => {
if (get(currentSession) !== sessionId || attempts >= MAX_RECONNECT) {
connectionState.set('disconnected')
streaming.set(false)
return
}
if (attempts > 0) {
connectionState.set('reconnecting')
addChatError(`Reconnecting to agent (attempt ${attempts + 1}/${MAX_RECONNECT})…`, 'Dismiss')
}
attempts++
const controller = streamChat(
'',
sessionId,
(_ev: ChatEvent) => {},
(_err: string) => {
delay = Math.min(delay * 2, 8000)
setTimeout(attemptReconnect, delay)
},
() => {
if (get(currentSession) === sessionId) {
connectionState.set('connected')
streaming.set(false)
loadSessionMessages(sessionId)
}
}
)
if (activeControllers.get(sessionId)) {
activeControllers.get(sessionId)?.abort()
}
activeControllers.set(sessionId, controller)
}
setTimeout(attemptReconnect, delay)
}
export function reconnect() {
const sid = get(currentSession)
if (!sid) return
connectionState.set('reconnecting')
const controller = streamChat(
'',
sid,
(_ev: ChatEvent) => {},
(_err: string) => {
connectionState.set('disconnected')
addChatError('Reconnect failed. The task may still be running — try sending a message to wake the agent.', 'Dismiss')
},
() => {
if (get(currentSession) === sid) {
connectionState.set('connected')
streaming.set(false)
loadSessionMessages(sid)
}
}
)
if (activeControllers.get(sid)) {
activeControllers.get(sid)?.abort()
}
activeControllers.set(sid, controller)
}
export function newChat() { export function newChat() {
cancelStream() cancelStream()
stopPolling() stopPolling()
connectionState.set('connected')
currentSession.set(null) currentSession.set(null)
messages.set([]) messages.set([])
error.set(null) error.set(null)
chatErrors.set([])
streaming.set(false) // fresh view — see loadSessionMessages for why this must not depend on cancelStream's own reset streaming.set(false) // fresh view — see loadSessionMessages for why this must not depend on cancelStream's own reset
} }

View File

@@ -1,5 +1,7 @@
<script lang="ts"> <script lang="ts">
import { messages, streaming, sendMessage, cancelStream, error } from '$lib/stores/chat' import { messages, streaming, connectionState, currentSession, sendMessage, cancelStream, reconnect, error, chatErrors, dismissError } from '$lib/stores/chat'
import { currentTask } from '$lib/stores/workspace'
import { resumeSession } from '$lib/api'
import SessionRail from '$lib/components/SessionRail.svelte' import SessionRail from '$lib/components/SessionRail.svelte'
import TaskContextPanel from '$lib/components/TaskContextPanel.svelte' import TaskContextPanel from '$lib/components/TaskContextPanel.svelte'
import ToolCallGroup from '$lib/components/ToolCallGroup.svelte' import ToolCallGroup from '$lib/components/ToolCallGroup.svelte'
@@ -8,7 +10,9 @@
import { Button } from '$lib/components/ui/button' import { Button } from '$lib/components/ui/button'
import { Textarea } from '$lib/components/ui/textarea' import { Textarea } from '$lib/components/ui/textarea'
import ArrowUpIcon from '@lucide/svelte/icons/arrow-up' import ArrowUpIcon from '@lucide/svelte/icons/arrow-up'
import RefreshCwIcon from '@lucide/svelte/icons/refresh-cw'
import SquareIcon from '@lucide/svelte/icons/square' import SquareIcon from '@lucide/svelte/icons/square'
import LoaderCircleIcon from '@lucide/svelte/icons/loader-circle'
import { marked } from 'marked' import { marked } from 'marked'
import DOMPurify from 'dompurify' import DOMPurify from 'dompurify'
@@ -94,6 +98,12 @@
if ($streaming) return if ($streaming) return
sendMessage(q) sendMessage(q)
} }
function statusLabel(s: string): string {
return s.replace(/_/g, ' ').replace(/\b\w/g, (c) => c.toUpperCase())
}
const liveStatus = $derived($currentTask?.status ?? ($currentSession ? 'active' : null))
</script> </script>
<div class="flex h-full min-h-0"> <div class="flex h-full min-h-0">
@@ -121,6 +131,30 @@
</div> </div>
{/if} {/if}
{#if $currentSession && $messages.length > 0}
<div class="flex items-center gap-2 text-xs text-muted-foreground">
{#if $streaming}
<LoaderCircleIcon class="size-3 shrink-0 animate-spin text-primary" />
<span>Agent is responding…</span>
{:else if liveStatus === 'executing'}
<LoaderCircleIcon class="size-3 shrink-0 animate-spin text-warning" />
<span>Working — {statusLabel(liveStatus)}</span>
<Button size="xs" variant="outline" class="ml-auto h-6 text-[11px]" onclick={async () => {
if ($currentSession) { await resumeSession($currentSession) }
}}>Continue</Button>
{:else if liveStatus === 'awaiting_input'}
<span class="text-warning">Waiting for your answer</span>
{:else if liveStatus}
<span>Status: {statusLabel(liveStatus)}</span>
{:else}
<span class="text-muted-foreground">Session ended</span>
{/if}
{#if $currentTask?.goal}
<span class="text-muted-foreground">· {$currentTask.goal.slice(0, 60)}{$currentTask.goal.length > 60 ? '…' : ''}</span>
{/if}
</div>
{/if}
{#each $messages as msg, i (msg.id)} {#each $messages as msg, i (msg.id)}
<div class="flex flex-col gap-1.5 {msg.role === 'user' ? 'items-end' : 'items-start'}"> <div class="flex flex-col gap-1.5 {msg.role === 'user' ? 'items-end' : 'items-start'}">
{#if msg.role === 'user'} {#if msg.role === 'user'}
@@ -161,6 +195,23 @@
</div> </div>
</div> </div>
{#if $connectionState === 'disconnected'}
<div class="mx-auto w-full max-w-3xl px-4">
<div class="mb-2 flex items-center gap-2 rounded-md border border-warning/50 bg-warning/10 px-3 py-2 text-xs">
<RefreshCwIcon class="size-3 shrink-0" aria-hidden="true" />
<span class="text-warning-foreground flex-1">Agent connection lost. The task may still be running.</span>
<Button size="xs" variant="outline" class="h-6 text-[11px]" onclick={reconnect}>Reconnect</Button>
</div>
</div>
{:else if $connectionState === 'reconnecting'}
<div class="mx-auto w-full max-w-3xl px-4">
<div class="mb-2 flex items-center gap-2 rounded-md border bg-muted/50 px-3 py-2 text-xs">
<RefreshCwIcon class="size-3 shrink-0 animate-spin text-muted-foreground" aria-hidden="true" />
<span class="text-muted-foreground flex-1">Reconnecting to agent…</span>
</div>
</div>
{/if}
{#if $error} {#if $error}
<div class="mx-auto w-full max-w-3xl px-4"> <div class="mx-auto w-full max-w-3xl px-4">
<div class="mb-2 rounded-md border border-destructive/50 bg-destructive/10 px-3 py-2 text-xs text-destructive"> <div class="mb-2 rounded-md border border-destructive/50 bg-destructive/10 px-3 py-2 text-xs text-destructive">
@@ -169,6 +220,18 @@
</div> </div>
{/if} {/if}
{#each $chatErrors as err (err.id)}
<div class="mx-auto w-full max-w-3xl px-4">
<div class="mb-2 flex items-center gap-2 rounded-md border border-destructive/30 bg-destructive/5 px-3 py-2 text-xs text-destructive">
<span class="flex-1">{err.message}</span>
{#if err.action}
<Button size="xs" variant="ghost" class="h-6 text-[11px]" onclick={() => dismissError(err.id)}>{err.action}</Button>
{/if}
<button class="ml-1 text-muted-foreground hover:text-foreground" onclick={() => dismissError(err.id)} aria-label="Dismiss">×</button>
</div>
</div>
{/each}
<div class="border-t bg-card/50 p-3"> <div class="border-t bg-card/50 p-3">
<form <form
class="mx-auto flex max-w-3xl items-end gap-2" class="mx-auto flex max-w-3xl items-end gap-2"