diff --git a/cmd/nomos/agent.go b/cmd/nomos/agent.go index cf1529f..0d69785 100644 --- a/cmd/nomos/agent.go +++ b/cmd/nomos/agent.go @@ -24,6 +24,20 @@ import ( const maxIterations = 40 const maxLLMRetries = 1 +// historyWindowSize bounds how many of a session's most recent persisted +// messages are replayed into the LLM's context on each turn — see +// store.go's getRecentMessages for why this exists (fix A2 of +// plans/2026-07-11-nomos-agent-code-review.md: unbounded history replay was +// a real, observed-in-production cost/latency/eventual-context-limit risk). +// 30 is a fixed-window choice, not token-budget-aware: simplest option that +// still keeps roughly the current task's working context, at the cost of +// occasionally dropping something a very long task still needed — the +// system note injected when truncation happens tells the model to check +// upsert_knowledge/search_knowledge rather than assume something didn't +// happen. A token-aware trim or LLM-summarize-on-drop are documented +// stretch options if a fixed window proves insufficient in practice. +const historyWindowSize = 30 + var refusalDenylist = []string{ "我没有相关信息", "您可以尝试问我其它问题", @@ -182,7 +196,15 @@ func (a *agent) chatWith(ctx context.Context, sessionID, message, systemInject s system += "\n\n" + snapshot } messages := []openai.ChatCompletionMessageParamUnion{openai.SystemMessage(system)} - history, _ := a.store.getMessages(ctx, sessionID) + history, truncatedHistory, _ := a.store.getRecentMessages(ctx, sessionID, historyWindowSize) + if truncatedHistory { + // Tell the model explicitly rather than silently dropping older + // turns — otherwise it might assume something wasn't done just + // because it doesn't see the turn that did it. + messages = append(messages, openai.SystemMessage(fmt.Sprintf( + "[System: this task has been running long enough that only the most recent %d turns of its history are included above your context — earlier turns happened but aren't shown. If you need to know what was already tried or found, check search_knowledge/get_entity_knowledge (if you recorded it) rather than assuming it didn't happen.]", + historyWindowSize))) + } var lastAssistantCalls []persistedCall for _, m := range history { text := extractText(m.Content) diff --git a/cmd/nomos/store.go b/cmd/nomos/store.go index e719618..8863cf4 100644 --- a/cmd/nomos/store.go +++ b/cmd/nomos/store.go @@ -215,6 +215,12 @@ func (s *store) listSessions(ctx context.Context) ([]session, error) { return out, rows.Err() } +// getMessages returns a session's ENTIRE message history, unbounded — used +// 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 +// it's run. For LLM replay, see getRecentMessages: sending the operator's +// full transcript is fine; sending the model's full transcript on every +// single turn is not (see getRecentMessages's doc comment). func (s *store) getMessages(ctx context.Context, sessionID string) ([]message, error) { if s == nil { return nil, nil @@ -238,6 +244,52 @@ func (s *store) getMessages(ctx context.Context, sessionID string) ([]message, e return out, rows.Err() } +// getRecentMessages returns the most recent `limit` messages for sessionID, +// in chronological order, plus whether older messages exist beyond that +// window. Used specifically for LLM replay (chatWith): without a bound, +// every turn re-sent the ENTIRE session history into the model's context, +// unconditionally growing with every turn — a real, observed-in-production +// cost/latency/eventual-context-limit risk for exactly the long-running, +// heavily-autonomous tasks (many auto-continuation cycles) this system is +// built to run longest. Fetches limit+1 rows to detect "there's more" +// without a separate COUNT query. +func (s *store) getRecentMessages(ctx context.Context, sessionID string, limit int) (msgs []message, truncated bool, err error) { + if s == nil { + return nil, false, nil + } + rows, qerr := s.pool.Query(ctx, + `SELECT id, session_id, role, content, created_at FROM agent_messages + WHERE session_id=$1 ORDER BY created_at DESC LIMIT $2`, + sessionID, limit+1) + if qerr != nil { + return nil, false, qerr + } + defer rows.Close() + + var out []message + for rows.Next() { + var m message + if err := rows.Scan(&m.ID, &m.SessionID, &m.Role, &m.Content, &m.CreatedAt); err != nil { + return nil, false, err + } + out = append(out, m) + } + if err := rows.Err(); err != nil { + return nil, false, err + } + + truncated = len(out) > limit + if truncated { + out = out[:limit] + } + // Rows came back newest-first (for the LIMIT to bound the right end); + // reverse to chronological order for replay. + for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 { + out[i], out[j] = out[j], out[i] + } + return out, truncated, nil +} + func (s *store) deleteSession(ctx context.Context, id string) error { if s == nil { return nil diff --git a/cmd/nomos/store_test.go b/cmd/nomos/store_test.go new file mode 100644 index 0000000..f0cb98c --- /dev/null +++ b/cmd/nomos/store_test.go @@ -0,0 +1,238 @@ +package main + +// Integration tests against a real Postgres, mirroring +// internal/db/integration_test.go's pattern: guarded by +// OIKOS_TEST_DATABASE_URL (skipped when unset), throwaway database per run, +// full migrations applied, dropped on cleanup. Run with: +// +// docker compose up -d postgres +// OIKOS_TEST_DATABASE_URL="postgres://oikos:oikos_dev@localhost:5432/oikos?sslmode=disable" go test ./cmd/nomos/ + +import ( + "context" + "fmt" + "math/rand" + "os" + "strings" + "testing" + + "github.com/dtoro/oikos/internal/db" + "github.com/jackc/pgx/v5" +) + +// newTestStore creates a throwaway, fully-migrated database and returns a +// *store connected to it, cleaned up (including a matching task: +// entity type in the ontology, needed by createTaskEntity/proposePlan tests) +// via t.Cleanup. +func newTestStore(t *testing.T) *store { + t.Helper() + baseURL := os.Getenv("OIKOS_TEST_DATABASE_URL") + if baseURL == "" { + t.Skip("OIKOS_TEST_DATABASE_URL not set — skipping integration test") + } + ctx := context.Background() + + admin, err := pgx.Connect(ctx, baseURL) + if err != nil { + t.Fatalf("connect admin: %v", err) + } + dbName := fmt.Sprintf("oikos_test_nomos_%08x", rand.Int63()) + if _, err := admin.Exec(ctx, "CREATE DATABASE "+dbName); err != nil { + admin.Close(ctx) + t.Fatalf("create test db: %v", err) + } + admin.Close(ctx) + + testURL := swapTestDatabase(baseURL, dbName) + pool, err := db.New(ctx, testURL) + if err != nil { + t.Fatalf("connect test db: %v", err) + } + t.Cleanup(func() { + pool.Close() + admin, err := pgx.Connect(ctx, baseURL) + if err == nil { + admin.Exec(ctx, "DROP DATABASE IF EXISTS "+dbName+" WITH (FORCE)") + admin.Close(ctx) + } + }) + if err := pool.Migrate(ctx); err != nil { + t.Fatalf("migrate: %v", err) + } + + // session_plan_steps/session_questions tests don't need the ontology + // seed, but createTaskEntity's INSERT INTO entities (type='task') has an + // FK to entity_types — seed the minimal rows it needs directly rather + // than pulling in the full seeds/ontology.yaml ingest path. + if _, err := pool.Exec(ctx, ` + INSERT INTO entity_types (name, domain, layer) VALUES ('entity', 'meta', 'meta') + ON CONFLICT DO NOTHING; + INSERT INTO entity_types (name, parent_type, domain, layer) VALUES ('task', 'entity', 'cognition', 'cognition') + ON CONFLICT DO NOTHING;`); err != nil { + t.Fatalf("seed minimal ontology: %v", err) + } + + return &store{pool: pool.Pool} +} + +func swapTestDatabase(url, dbName string) string { + qi := strings.Index(url, "?") + params, base := "", url + if qi >= 0 { + params = url[qi:] + base = url[:qi] + } + si := strings.LastIndex(base, "/") + return base[:si+1] + dbName + params +} + +// TestGetRecentMessages_Truncation is the concrete proof for fix A2 of +// plans/2026-07-11-nomos-agent-code-review.md: chatWith used to replay a +// session's ENTIRE history on every turn with no bound. getRecentMessages +// caps that; this test checks both sides — under the limit, nothing is +// dropped and truncated=false; over it, only the most recent `limit` come +// back, in chronological order, with truncated=true. +func TestGetRecentMessages_Truncation(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + sess, err := s.createSession(ctx, "history window test") + if err != nil { + t.Fatalf("createSession: %v", err) + } + + const total = 35 + const limit = 30 + for i := 0; i < total; i++ { + role := "user" + if i%2 == 1 { + role = "assistant" + } + body := fmt.Appendf(nil, `{"role":%q,"text":"msg-%d"}`, role, i) + if err := s.saveMessage(ctx, sess.ID, role, body); err != nil { + t.Fatalf("saveMessage %d: %v", i, err) + } + } + + msgs, truncated, err := s.getRecentMessages(ctx, sess.ID, limit) + if err != nil { + t.Fatalf("getRecentMessages: %v", err) + } + if !truncated { + t.Errorf("truncated = false, want true (%d messages > limit %d)", total, limit) + } + if len(msgs) != limit { + t.Fatalf("got %d messages, want %d", len(msgs), limit) + } + // Chronological order: the oldest of the RETAINED messages should be the + // (total-limit)-th one saved (msg-5, since msg-0..4 were dropped), and + // the last should be the most recently saved (msg-34). + wantFirst := fmt.Sprintf("msg-%d", total-limit) + wantLast := fmt.Sprintf("msg-%d", total-1) + if got := extractText(msgs[0].Content); got != wantFirst { + t.Errorf("first retained message = %q, want %q", got, wantFirst) + } + if got := extractText(msgs[len(msgs)-1].Content); got != wantLast { + t.Errorf("last retained message = %q, want %q", got, wantLast) + } + + // Under the limit: nothing dropped. + sess2, err := s.createSession(ctx, "small session") + if err != nil { + t.Fatalf("createSession: %v", err) + } + for i := 0; i < 5; i++ { + body := fmt.Appendf(nil, `{"role":"user","text":"msg-%d"}`, i) + if err := s.saveMessage(ctx, sess2.ID, "user", body); err != nil { + t.Fatalf("saveMessage: %v", err) + } + } + msgs2, truncated2, err := s.getRecentMessages(ctx, sess2.ID, limit) + if err != nil { + t.Fatalf("getRecentMessages (small): %v", err) + } + if truncated2 { + t.Errorf("truncated = true for a 5-message session under a %d limit, want false", limit) + } + if len(msgs2) != 5 { + t.Errorf("got %d messages, want 5", len(msgs2)) + } +} + +// TestProposePlan_AppendVsReplace is the concrete proof for the plan-append +// fix (commit 5384499, "plan panel showed only the latest step, not the full +// plan"): proposePlan must REPLACE the step list only while every existing +// step is still 'pending' (a genuine pre-execution revision), and APPEND +// once any step has started — otherwise a model that calls propose_plan once +// per step (rather than once with the full list, as instructed) erases every +// already-completed step each time, and the operator only ever sees the +// latest single step instead of real progress. +func TestProposePlan_AppendVsReplace(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + sess, err := s.createSession(ctx, "plan append test") + if err != nil { + t.Fatalf("createSession: %v", err) + } + + // First call: no steps exist yet — must persist as-is (replace mode, + // trivially: nothing to replace). + out1, err := s.proposePlan(ctx, sess.ID, []planStepInput{{Title: "Step A"}}) + if err != nil { + t.Fatalf("proposePlan #1: %v", err) + } + if len(out1) != 1 || out1[0]["seq"] != 1 { + t.Fatalf("proposePlan #1 = %+v, want one step at seq 1", out1) + } + + // Mark step 1 as started. + if err := s.updatePlanStep(ctx, sess.ID, 1, "running", ""); err != nil { + t.Fatalf("updatePlanStep: %v", err) + } + + // Second call, simulating a model that (against instructions) calls + // propose_plan again per-step instead of once with the full list: since + // step 1 has left 'pending', this MUST append, not replace. + out2, err := s.proposePlan(ctx, sess.ID, []planStepInput{{Title: "Step B"}}) + if err != nil { + t.Fatalf("proposePlan #2: %v", err) + } + if len(out2) != 1 || out2[0]["seq"] != 2 { + t.Fatalf("proposePlan #2 = %+v, want one step at seq 2 (appended after the running step 1)", out2) + } + + steps, err := s.getPlanSteps(ctx, sess.ID) + if err != nil { + t.Fatalf("getPlanSteps: %v", err) + } + if len(steps) != 2 { + t.Fatalf("got %d persisted steps, want 2 (step 1 must survive the second propose_plan call)", len(steps)) + } + if steps[0].Title != "Step A" || steps[0].Status != "running" { + t.Errorf("step 1 = %+v, want Step A still running (not erased)", steps[0]) + } + if steps[1].Title != "Step B" || steps[1].Status != "pending" { + t.Errorf("step 2 = %+v, want Step B pending", steps[1]) + } + + // Third call BEFORE anything runs on a fresh session: every step is + // still pending, so this must REPLACE, not append. + sess2, err := s.createSession(ctx, "plan replace test") + if err != nil { + t.Fatalf("createSession: %v", err) + } + if _, err := s.proposePlan(ctx, sess2.ID, []planStepInput{{Title: "Original"}}); err != nil { + t.Fatalf("proposePlan (initial): %v", err) + } + if _, err := s.proposePlan(ctx, sess2.ID, []planStepInput{{Title: "Revised"}}); err != nil { + t.Fatalf("proposePlan (revise before execution): %v", err) + } + revisedSteps, err := s.getPlanSteps(ctx, sess2.ID) + if err != nil { + t.Fatalf("getPlanSteps: %v", err) + } + if len(revisedSteps) != 1 || revisedSteps[0].Title != "Revised" { + t.Fatalf("got %+v, want a single 'Revised' step (pre-execution revise must replace, not append)", revisedSteps) + } +}