fix(agent): bound conversation history replayed to the LLM (A2)

Fix A2 of plans/2026-07-11-nomos-agent-code-review.md. chatWith replayed a
session's ENTIRE message history into the LLM's context on EVERY turn, no
windowing, no token budget — confirmed against a documented production case
(a single turn with 70 tool calls, messages up to 106KB). Every subsequent
turn of a long-running or heavily-autonomous task re-sent that ever-growing
history in full — a real cost/latency/eventual-context-limit risk for
exactly the tasks this system runs longest (many auto-continuation cycles).

Design call (flagged in the review as needing one before implementation):
a fixed-size window for LLM replay specifically, not the UI's own transcript
view. Simplest option that still keeps roughly the current task's working
context; a token-aware trim or LLM-summarize-on-drop are documented as
stretch options if 30 proves insufficient in practice.

- store.go: new getRecentMessages(ctx, sessionID, limit) — last `limit`
  messages in chronological order, plus whether older ones were omitted.
  getMessages (used by the UI's GET /sessions/{id}) is untouched and stays
  unbounded — the operator should still see a task's full history regardless
  of length; only what gets sent to the model is bounded.
- agent.go: chatWith uses getRecentMessages(sessionID, historyWindowSize=30)
  instead of the unbounded getMessages. When truncated, injects a system
  note telling the model explicitly that older turns exist but aren't shown,
  so it checks upsert_knowledge/search_knowledge rather than assuming
  something wasn't done just because it isn't visible.

New cmd/nomos/store_test.go: real Postgres integration tests (mirroring
internal/db/integration_test.go's throwaway-database pattern, guarded by
OIKOS_TEST_DATABASE_URL). TestGetRecentMessages_Truncation is the direct
proof for this fix (35 messages → 30 returned, correctly ordered,
truncated=true; 5 messages → all 5, truncated=false) — both cases run
against a fully-migrated database, not mocked. Also added
TestProposePlan_AppendVsReplace, closing part of the review's test-coverage
finding (E) by permanently regression-testing the earlier append-vs-replace
plan fix (commit 5384499), which had only been verified manually until now.

Verified live: inflated a real session to 42 persisted messages via direct
SQL, then continued it with a real chat call — the turn proceeded normally
(multiple real tool-call iterations, no crash, no context-length error);
nomos stayed healthy throughout. A3's incremental persistence separately
confirmed to have caught the 7 real tool calls made before the client
connection was cut, cleanly closing out both fixes' interaction.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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)

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

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:<session>
+// 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)
+	}
+}