diff --git a/apps/backend/internal/office/models/continuation_scope.go b/apps/backend/internal/office/models/continuation_scope.go new file mode 100644 index 0000000000..41286665d3 --- /dev/null +++ b/apps/backend/internal/office/models/continuation_scope.go @@ -0,0 +1,46 @@ +package models + +import "encoding/json" + +// ContinuationScopeForRun computes the continuation-summary scope key for +// a run: "routine:" when the run's context snapshot carries a +// routine_id (set by the wakeup dispatcher for source="routine" wakeups), +// else "agent:" so non-routine fires still have a stable upsert +// key. The legacy "heartbeat" scope is retired alongside the agent-level +// heartbeat cron. +// +// Callers must not invoke this more than once per run. It is computed a +// single time at run-creation (runs/repository/sqlite.CreateRun) and +// persisted on Run.ContinuationScope; every later reader or writer reads +// that stored value instead of re-deriving it. A routine wakeup that +// coalesces into an already-claimed run patches only context_snapshot +// (see wakeup.Dispatcher / MarkWakeupRequestCoalesced), so a second +// derivation against a freshly re-fetched row can disagree with the +// derivation a claim-time in-memory copy is still holding — sharing this +// function is not enough on its own when its input can drift between +// calls. +func ContinuationScopeForRun(run *Run, agentProfileID string) string { + if run == nil { + return "agent:" + agentProfileID + } + if id := extractRoutineID(run.ContextSnapshot); id != "" { + return "routine:" + id + } + return "agent:" + agentProfileID +} + +// extractRoutineID pulls routine_id out of a JSON context snapshot. +// Returns "" for missing / malformed payloads so the caller falls back to +// the agent-scoped key. +func extractRoutineID(snapshot string) string { + if snapshot == "" { + return "" + } + var p struct { + RoutineID string `json:"routine_id"` + } + if err := json.Unmarshal([]byte(snapshot), &p); err != nil { + return "" + } + return p.RoutineID +} diff --git a/apps/backend/internal/office/models/continuation_scope_test.go b/apps/backend/internal/office/models/continuation_scope_test.go new file mode 100644 index 0000000000..b35df3c73e --- /dev/null +++ b/apps/backend/internal/office/models/continuation_scope_test.go @@ -0,0 +1,44 @@ +package models + +import "testing" + +func TestContinuationScopeForRun(t *testing.T) { + tests := []struct { + name string + run *Run + agentProfileID string + want string + }{ + { + name: "routine scope", + run: &Run{ContextSnapshot: `{"routine_id":"routine-1"}`}, + agentProfileID: "agent-1", + want: "routine:routine-1", + }, + { + name: "agent fallback", + run: &Run{ContextSnapshot: `{}`}, + agentProfileID: "agent-1", + want: "agent:agent-1", + }, + { + name: "malformed snapshot fallback", + run: &Run{ContextSnapshot: "not-json"}, + agentProfileID: "agent-1", + want: "agent:agent-1", + }, + { + name: "nil run fallback", + agentProfileID: "agent-1", + want: "agent:agent-1", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := ContinuationScopeForRun(tt.run, tt.agentProfileID); got != tt.want { + t.Fatalf("ContinuationScopeForRun() = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/apps/backend/internal/office/models/models.go b/apps/backend/internal/office/models/models.go index 24f9d39df1..af6b4e4a60 100644 --- a/apps/backend/internal/office/models/models.go +++ b/apps/backend/internal/office/models/models.go @@ -410,10 +410,20 @@ type Run struct { // SummaryInjected is the continuation-summary content prepended // to the prompt at dispatch time, snapshot for inspection. Empty // when no summary was injected (today: every run, until PR 2). - SummaryInjected string `json:"summary_injected,omitempty" db:"summary_injected"` - RequestedAt time.Time `json:"requested_at" db:"requested_at"` - ClaimedAt *time.Time `json:"claimed_at" db:"claimed_at"` - FinishedAt *time.Time `json:"finished_at" db:"finished_at"` + SummaryInjected string `json:"summary_injected,omitempty" db:"summary_injected"` + // ContinuationScope is the continuation-summary scope key + // (ContinuationScopeForRun's output) computed once at run creation + // and persisted here so every later reader/writer of this run's + // continuation summary uses the same value. Computing this at + // creation time — before any wakeup can coalesce into this row — + // closes a race where a routine wakeup patches context_snapshot + // after claim but a re-derivation against the freshly re-fetched + // row would disagree with the derivation the claiming scheduler is + // still holding in memory. + ContinuationScope string `json:"continuation_scope,omitempty" db:"continuation_scope"` + RequestedAt time.Time `json:"requested_at" db:"requested_at"` + ClaimedAt *time.Time `json:"claimed_at" db:"claimed_at"` + FinishedAt *time.Time `json:"finished_at" db:"finished_at"` // Provider-routing columns (office-provider-routing spec). All // optional and ignored when workspace routing is disabled. The TEXT diff --git a/apps/backend/internal/office/repository/sqlite/base.go b/apps/backend/internal/office/repository/sqlite/base.go index d144e73f16..8501617ffc 100644 --- a/apps/backend/internal/office/repository/sqlite/base.go +++ b/apps/backend/internal/office/repository/sqlite/base.go @@ -363,6 +363,13 @@ func (r *Repository) createRunTables() error { result_json TEXT NOT NULL DEFAULT '{}', assembled_prompt TEXT NOT NULL DEFAULT '', summary_injected TEXT NOT NULL DEFAULT '', + -- continuation_scope is the continuation-summary scope key + -- (models.ContinuationScopeForRun) decided once at run creation + -- and persisted so every later reader/writer of this run's + -- continuation summary uses the same value instead of + -- re-deriving it against a context_snapshot a coalesced wakeup + -- may have since patched. + continuation_scope TEXT NOT NULL DEFAULT '', requested_at TIMESTAMP NOT NULL, claimed_at TIMESTAMP, finished_at TIMESTAMP diff --git a/apps/backend/internal/office/repository/sqlite/base_migrations.go b/apps/backend/internal/office/repository/sqlite/base_migrations.go index 8be50139a5..4867dc987c 100644 --- a/apps/backend/internal/office/repository/sqlite/base_migrations.go +++ b/apps/backend/internal/office/repository/sqlite/base_migrations.go @@ -7,6 +7,8 @@ import ( "strings" "go.uber.org/zap" + + "github.com/kandev/kandev/internal/office/models" ) // runMigrations applies idempotent schema migrations for office tables @@ -36,6 +38,48 @@ func (r *Repository) runMigrations() { // Provider routing tables and replayable column migrations. Fresh // schemas include the columns inline; ALTERs converge existing databases. r.migrateProviderRouting() + r.migrateContinuationScope() +} + +// migrateContinuationScope adds runs.continuation_scope for databases +// created before WO-16's claim-time scope persistence. Existing rows receive +// a scope from their stored context snapshot so queued or claimed taskless +// runs keep their continuation chain after an upgrade. +func (r *Repository) migrateContinuationScope() { + r.migrate.Apply("runs.continuation_scope", + `ALTER TABLE runs ADD COLUMN continuation_scope TEXT NOT NULL DEFAULT ''`) + r.backfillContinuationScopes() +} + +func (r *Repository) backfillContinuationScopes() { + var legacyRuns []struct { + ID string `db:"id"` + AgentProfileID string `db:"agent_profile_id"` + ContextSnapshot string `db:"context_snapshot"` + } + if err := r.db.Select(&legacyRuns, + `SELECT id, agent_profile_id, context_snapshot + FROM runs WHERE continuation_scope = ''`); err != nil { + if r.log != nil { + r.log.Warn("continuation scope backfill query failed", zap.Error(err)) + } + return + } + for _, legacyRun := range legacyRuns { + scope := models.ContinuationScopeForRun( + &models.Run{ContextSnapshot: legacyRun.ContextSnapshot}, + legacyRun.AgentProfileID, + ) + if _, err := r.db.Exec(r.db.Rebind(` + UPDATE runs SET continuation_scope = ? + WHERE id = ? AND continuation_scope = '' + `), scope, legacyRun.ID); err != nil { + if r.log != nil { + r.log.Warn("continuation scope backfill update failed", + zap.String("run_id", legacyRun.ID), zap.Error(err)) + } + } + } } // migrateCostEventContract adds the cache read/write split, turn diff --git a/apps/backend/internal/office/repository/sqlite/continuation_scope_migration_postgres_test.go b/apps/backend/internal/office/repository/sqlite/continuation_scope_migration_postgres_test.go new file mode 100644 index 0000000000..797b2b749c --- /dev/null +++ b/apps/backend/internal/office/repository/sqlite/continuation_scope_migration_postgres_test.go @@ -0,0 +1,67 @@ +package sqlite_test + +import ( + "context" + "testing" + + "github.com/jmoiron/sqlx" + + "github.com/kandev/kandev/internal/office/repository/sqlite" + taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite" + "github.com/kandev/kandev/internal/testutil" +) + +// TestPostgresContinuationScopeMigration is the PostgreSQL twin of +// TestContinuationScopeMigrationBackfillsLegacyRunsAndReplays. It verifies +// that an upgraded database repairs queued and claimed runs, then replays the +// migration cleanly on the next boot. It skips unless a test DSN is set. +func TestPostgresContinuationScopeMigration(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + ctx := context.Background() + + if _, err := taskrepo.NewWithDB(db, db, nil); err != nil { + t.Fatalf("init task repo: %v", err) + } + if _, err := sqlite.NewWithDB(db, db, nil); err != nil { + t.Fatalf("init current office schema: %v", err) + } + + if _, err := db.ExecContext(ctx, `ALTER TABLE runs DROP COLUMN continuation_scope`); err != nil { + t.Fatalf("drop continuation_scope from legacy schema: %v", err) + } + if _, err := db.ExecContext(ctx, ` + INSERT INTO runs ( + id, agent_profile_id, reason, payload, status, context_snapshot, requested_at + ) VALUES + ('legacy-routine', 'agent-routine', 'routine_trigger', '{}', 'queued', + '{"routine_id":"routine-1"}', CURRENT_TIMESTAMP), + ('legacy-agent', 'agent-plain', 'self', '{}', 'claimed', + '{}', CURRENT_TIMESTAMP) + `); err != nil { + t.Fatalf("seed legacy runs: %v", err) + } + + if _, err := sqlite.NewWithDB(db, db, nil); err != nil { + t.Fatalf("migrate legacy schema: %v", err) + } + assertContinuationScopePostgres(t, db, "legacy-routine", "routine:routine-1") + assertContinuationScopePostgres(t, db, "legacy-agent", "agent:agent-plain") + + if _, err := sqlite.NewWithDB(db, db, nil); err != nil { + t.Fatalf("replay migration: %v", err) + } + assertContinuationScopePostgres(t, db, "legacy-routine", "routine:routine-1") + assertContinuationScopePostgres(t, db, "legacy-agent", "agent:agent-plain") +} + +func assertContinuationScopePostgres(t *testing.T, db *sqlx.DB, runID, want string) { + t.Helper() + var got string + if err := db.QueryRowContext(context.Background(), + `SELECT continuation_scope FROM runs WHERE id = $1`, runID).Scan(&got); err != nil { + t.Fatalf("read continuation scope for %q: %v", runID, err) + } + if got != want { + t.Errorf("continuation_scope for %q = %q, want %q", runID, got, want) + } +} diff --git a/apps/backend/internal/office/repository/sqlite/continuation_scope_migration_test.go b/apps/backend/internal/office/repository/sqlite/continuation_scope_migration_test.go new file mode 100644 index 0000000000..c70f055d47 --- /dev/null +++ b/apps/backend/internal/office/repository/sqlite/continuation_scope_migration_test.go @@ -0,0 +1,60 @@ +package sqlite_test + +import ( + "testing" + + "github.com/jmoiron/sqlx" + _ "github.com/mattn/go-sqlite3" + + "github.com/kandev/kandev/internal/office/repository/sqlite" +) + +func TestContinuationScopeMigrationBackfillsLegacyRunsAndReplays(t *testing.T) { + db, err := sqlx.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + db.SetMaxOpenConns(1) + t.Cleanup(func() { _ = db.Close() }) + + if _, err := sqlite.NewWithDB(db, db, nil); err != nil { + t.Fatalf("create current schema: %v", err) + } + if _, err := db.Exec(`ALTER TABLE runs DROP COLUMN continuation_scope`); err != nil { + t.Fatalf("drop continuation_scope from legacy schema: %v", err) + } + if _, err := db.Exec(` + INSERT INTO runs ( + id, agent_profile_id, reason, payload, status, context_snapshot, requested_at + ) VALUES + ('legacy-routine', 'agent-routine', 'routine_trigger', '{}', 'queued', + '{"routine_id":"routine-1"}', CURRENT_TIMESTAMP), + ('legacy-agent', 'agent-plain', 'self', '{}', 'claimed', + '{}', CURRENT_TIMESTAMP) + `); err != nil { + t.Fatalf("seed legacy runs: %v", err) + } + + if _, err := sqlite.NewWithDB(db, db, nil); err != nil { + t.Fatalf("migrate legacy schema: %v", err) + } + assertContinuationScope(t, db, "legacy-routine", "routine:routine-1") + assertContinuationScope(t, db, "legacy-agent", "agent:agent-plain") + + if _, err := sqlite.NewWithDB(db, db, nil); err != nil { + t.Fatalf("replay migration: %v", err) + } + assertContinuationScope(t, db, "legacy-routine", "routine:routine-1") + assertContinuationScope(t, db, "legacy-agent", "agent:agent-plain") +} + +func assertContinuationScope(t *testing.T, db *sqlx.DB, runID, want string) { + t.Helper() + var got string + if err := db.Get(&got, `SELECT continuation_scope FROM runs WHERE id = ?`, runID); err != nil { + t.Fatalf("read continuation scope for %q: %v", runID, err) + } + if got != want { + t.Errorf("continuation_scope for %q = %q, want %q", runID, got, want) + } +} diff --git a/apps/backend/internal/office/service/continuation_summary_reader_test.go b/apps/backend/internal/office/service/continuation_summary_reader_test.go new file mode 100644 index 0000000000..cd7329237c --- /dev/null +++ b/apps/backend/internal/office/service/continuation_summary_reader_test.go @@ -0,0 +1,205 @@ +package service_test + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/office/models" + "github.com/kandev/kandev/internal/office/service" +) + +// TestLoadContinuationSummary_AgentScope_RoundTripsThroughRealWriterAndReader +// pins WO-16: the reader must key off run.ContinuationScope, the value +// models.ContinuationScopeForRun computed once at run creation and that +// the writer reads back at completion — not the retired "heartbeat" +// literal, and not a fresh re-derivation. A taskless run with no +// routine_id in its context_snapshot is written under "agent:" +// (ContinuationScopeForRun's fallback branch); the reader must find that +// same row when assembling the prompt for the next taskless wake. +func TestLoadContinuationSummary_AgentScope_RoundTripsThroughRealWriterAndReader(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + ensureTaskSessionMessagesTable(t, svc) + + createTestAgent(t, svc, "ws-1", "agent-scope-a") + + if err := svc.QueueRun( + ctx, "agent-scope-a", service.RunReasonTaskAssigned, "{}", "continuation-agent-scope", + ); err != nil { + t.Fatalf("queue run: %v", err) + } + run, err := svc.ClaimNextRun(ctx) + if err != nil || run == nil { + t.Fatalf("claim run: %v (run=%v)", err, run) + } + + completed := bus.NewEvent(events.AgentCompleted, "test", map[string]string{ + "agent_id": "claude-acp", + "agent_profile_id": "agent-scope-a", + "session_id": "sess-scope-a", + }) + if pErr := eb.Publish(ctx, events.AgentCompleted, completed); pErr != nil { + t.Fatalf("publish completed: %v", pErr) + } + + // Confirm the writer actually landed a row under "agent:agent-scope-a" + // before exercising the reader, so a reader failure below is + // unambiguously the reader's bug and not a writer regression. + written, err := svc.GetContinuationSummaryForTest(ctx, "agent-scope-a", "agent:agent-scope-a") + if err != nil || written == nil || written.Content == "" { + t.Fatalf("writer did not produce a summary under agent:agent-scope-a: %v (row=%v)", err, written) + } + + // ContinuationScope is set explicitly here, as it would be by a real + // ClaimNextRun-returned run: CreateRun computes and persists it once, + // at creation time, and the reader now reads that stored value + // directly instead of re-deriving it from ContextSnapshot. + nextRun := &models.Run{ContextSnapshot: "{}", ContinuationScope: "agent:agent-scope-a"} + si := service.NewSchedulerIntegration(svc, time.Minute) + got := si.LoadContinuationSummaryForTest(ctx, nextRun, "agent-scope-a", "") + if got == "" { + t.Fatalf(`loadContinuationSummary returned "" for the agent: scope; ` + + `reader must key off run.ContinuationScope, not the retired "heartbeat" literal`) + } + if got != written.Content { + t.Errorf("loadContinuationSummary = %q, want writer's content %q", got, written.Content) + } +} + +// TestLoadContinuationSummary_RoutineScope_RoundTripsThroughRealWriterAndReader +// covers the writer's other branch: a taskless run dispatched from a +// routine carries routine_id in its context_snapshot, so +// ContinuationScopeForRun keys the upsert "routine:" instead of +// "agent:". This is the case a naive `"agent:"+agentID` reader fix +// would still leave broken — see WO-16 Decision 2. The row is seeded +// with continuation_scope already populated, simulating a run that went +// through the real CreateRun path (which decides the scope once, at +// creation, before this row could ever be coalesced into). +func TestLoadContinuationSummary_RoutineScope_RoundTripsThroughRealWriterAndReader(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + ensureTaskSessionMessagesTable(t, svc) + + createTestAgent(t, svc, "ws-1", "agent-scope-r") + + claimedAt := time.Now().UTC() + svc.ExecSQL(t, ` + INSERT INTO runs ( + id, agent_profile_id, reason, payload, status, coalesced_count, + context_snapshot, continuation_scope, requested_at, claimed_at + ) VALUES ( + 'run-scope-r', 'agent-scope-r', 'routine_trigger', '{}', + 'claimed', 1, '{"routine_id":"rt-1"}', 'routine:rt-1', ?, ? + ) + `, claimedAt, claimedAt) + + completed := bus.NewEvent(events.AgentCompleted, "test", map[string]string{ + "agent_id": "claude-acp", + "agent_profile_id": "agent-scope-r", + "run_id": "run-scope-r", + "session_id": "sess-scope-r", + }) + if pErr := eb.Publish(ctx, events.AgentCompleted, completed); pErr != nil { + t.Fatalf("publish completed: %v", pErr) + } + + written, err := svc.GetContinuationSummaryForTest(ctx, "agent-scope-r", "routine:rt-1") + if err != nil || written == nil || written.Content == "" { + t.Fatalf("writer did not produce a summary under routine:rt-1: %v (row=%v)", err, written) + } + + nextRun := &models.Run{ContextSnapshot: `{"routine_id":"rt-1"}`, ContinuationScope: "routine:rt-1"} + si := service.NewSchedulerIntegration(svc, time.Minute) + got := si.LoadContinuationSummaryForTest(ctx, nextRun, "agent-scope-r", "") + if got == "" { + t.Fatalf(`loadContinuationSummary returned "" for the routine: scope; ` + + `a reader hardcoded to "agent:"+agentID would fail exactly this case`) + } + if got != written.Content { + t.Errorf("loadContinuationSummary = %q, want writer's content %q", got, written.Content) + } +} + +// TestLoadContinuationSummary_ScopeSurvivesCoalesceAfterClaim pins the PR +// #2971 review round-2 finding: reader and writer must agree on scope +// even when a routine wakeup coalesces into an already-claimed run +// between claim and completion. The reader holds the claimed run in memory, +// while the writer fetches it again after completion. Coalescing patches +// context_snapshot, so re-deriving the scope would let the two paths select +// different summary rows. CreateRun persists ContinuationScope before a row +// can be coalesced, so both paths use the same bucket. +// +// This run originates from a plain (non-routine) taskless wakeup, so its +// scope is decided as "agent:" at creation, before a routine wakeup +// coalesces into it — exactly the ordering wakeup.Dispatcher.Dispatch +// produces when a routine fires for an agent that already has an +// in-flight run (PolicyCoalesceIfActive). On pre-fix code this test +// fails: the writer lands the summary under "routine:rt-2" instead, so +// the final read below (using the claim-time run, still scoped +// "agent:") comes back empty. +func TestLoadContinuationSummary_ScopeSurvivesCoalesceAfterClaim(t *testing.T) { + svc, eb := newTestServiceWithBus(t) + ctx := context.Background() + ensureTaskSessionMessagesTable(t, svc) + + createTestAgent(t, svc, "ws-1", "agent-scope-coalesce") + + if err := svc.QueueRun( + ctx, "agent-scope-coalesce", service.RunReasonTaskAssigned, "{}", "continuation-scope-coalesce", + ); err != nil { + t.Fatalf("queue run: %v", err) + } + // claimedRun is the scheduler's claim-time snapshot — the same + // in-memory object assembleAgentPrompt's reader uses. It is captured + // once, here, and never re-read from the DB below, mirroring + // processRun's real control flow (prepareAndLaunch carries this same + // object through to assembleAgentPrompt). + claimedRun, err := svc.ClaimNextRun(ctx) + if err != nil || claimedRun == nil { + t.Fatalf("claim run: %v (run=%v)", err, claimedRun) + } + if claimedRun.ContinuationScope != "agent:agent-scope-coalesce" { + t.Fatalf("claimed run continuation_scope = %q, want %q", + claimedRun.ContinuationScope, "agent:agent-scope-coalesce") + } + + // A routine wakeup fires for the same agent while this run is still + // in flight and coalesces into it — the same call the wakeup + // dispatcher makes when PolicyCoalesceIfActive finds an in-flight run. + svc.CoalesceRoutineWakeupForTest(ctx, t, "agent-scope-coalesce", claimedRun.ID, "rt-2") + + completed := bus.NewEvent(events.AgentCompleted, "test", map[string]string{ + "agent_id": "claude-acp", + "agent_profile_id": "agent-scope-coalesce", + "run_id": claimedRun.ID, + "session_id": "sess-scope-coalesce", + }) + if pErr := eb.Publish(ctx, events.AgentCompleted, completed); pErr != nil { + t.Fatalf("publish completed: %v", pErr) + } + + // The writer must have landed the summary under the run's persisted + // creation-time scope, unaffected by the coalesce. + written, err := svc.GetContinuationSummaryForTest(ctx, "agent-scope-coalesce", "agent:agent-scope-coalesce") + if err != nil || written == nil || written.Content == "" { + t.Fatalf( + "writer did not produce a summary under agent:agent-scope-coalesce after a "+ + "post-claim routine coalesce: %v (row=%v)", err, written) + } + + // The reader, using the exact claim-time run object (unaware of the + // later coalesce), must find that same row. + si := service.NewSchedulerIntegration(svc, time.Minute) + got := si.LoadContinuationSummaryForTest(ctx, claimedRun, "agent-scope-coalesce", "") + if got == "" { + t.Fatal( + "loadContinuationSummary returned \"\" after a routine wakeup coalesced into an " + + "already-claimed run; reader and writer disagreed on scope") + } + if got != written.Content { + t.Errorf("loadContinuationSummary = %q, want writer's content %q", got, written.Content) + } +} diff --git a/apps/backend/internal/office/service/event_subscribers.go b/apps/backend/internal/office/service/event_subscribers.go index 460d61aac8..04106527bc 100644 --- a/apps/backend/internal/office/service/event_subscribers.go +++ b/apps/backend/internal/office/service/event_subscribers.go @@ -94,22 +94,15 @@ type ApprovalResolvedData struct { // AgentLifecycleData is the subset of agent lifecycle event data needed by office. // -// AgentID is populated by the lifecycle manager for taskless runs -// (heartbeats, lightweight routines) so the office completion handler -// can attribute the run without a task lookup. It stays empty for the -// task-bound path that already uses TaskID. Today no caller emits -// taskless lifecycle events; the field is reserved for PR 2 of -// office-heartbeat-rework. Note that AgentID carries the underlying agent -// TYPE (e.g. "claude-acp"), not an office agent identity — it is unrelated -// to AgentProfileID below despite the similar name. +// AgentID is the underlying agent type (for example, "claude-acp"). It is +// not an Office agent identity and must not be used to look up runs or +// continuation summaries. // // AgentProfileID is the office agent instance's own identity // (lifecycle.AgentEventPayload's "agent_profile_id", sourced from -// AgentExecution.officeProfileID()) and is always populated on the -// task-bound path. runs.agent_profile_id is keyed on this same identity, so -// callers that need to resolve the specific run a lifecycle event belongs -// to — as opposed to "some claimed run on this task" — must match on this -// field, not AgentID (Review round 4, BLOCKING FINDING 2). +// AgentExecution.officeProfileID()). runs.agent_profile_id is keyed on this +// same identity, so taskless fallback resolution and summary writes use this +// field, not AgentID. type AgentLifecycleData struct { TaskID string `json:"task_id"` RunID string `json:"run_id"` @@ -142,7 +135,12 @@ func (s *Service) resolveLifecycleRun(ctx context.Context, data AgentLifecycleDa if data.TaskID != "" { return s.repo.GetClaimedRunByTaskAndAgent(ctx, data.TaskID, data.AgentProfileID) } - return s.repo.GetClaimedTasklessRunForAgent(ctx, data.AgentID) + agentProfileID := data.AgentProfileID + if agentProfileID == "" { + // Legacy taskless events used AgentID for the Office identity. + agentProfileID = data.AgentID + } + return s.repo.GetClaimedTasklessRunForAgent(ctx, agentProfileID) } type PromptUsageData struct { @@ -330,10 +328,8 @@ func (s *Service) handleAgentCompleted(ctx context.Context, event *bus.Event) er if err != nil { return nil } - // Taskless completion (PR 1 of office-heartbeat-rework): heartbeat - // or lightweight-routine fires that don't carry a task_id. Today no - // caller emits these, so this branch is dead until PR 2 lands the - // agent_heartbeat cron handler. + // Taskless completion from a routine or other wakeup does not carry a + // task_id, so resolve it by run ID or Office agent profile. if data.TaskID == "" { return s.handleTasklessAgentCompleted(ctx, data) } @@ -386,8 +382,8 @@ func (s *Service) markRoutingSuccess(ctx context.Context, run *models.Run) { } // handleTasklessAgentCompleted attributes a taskless run completion, -// finishes the run, and refreshes the per-(agent, "heartbeat") -// continuation summary so the next fire has bridge context. The +// finishes the run, and refreshes the per-agent, per-scope continuation +// summary so the next fire has bridge context. The // summary is built deterministically from the run's result_json, // workspace activity, and the prior summary — see the office/summary // package. @@ -398,7 +394,7 @@ func (s *Service) markRoutingSuccess(ctx context.Context, run *models.Run) { func (s *Service) handleTasklessAgentCompleted( ctx context.Context, data *AgentLifecycleData, ) error { - if data == nil || (data.AgentID == "" && data.RunID == "") { + if data == nil || (data.AgentID == "" && data.AgentProfileID == "" && data.RunID == "") { return nil } run, err := s.resolveLifecycleRun(ctx, *data) @@ -412,7 +408,7 @@ func (s *Service) handleTasklessAgentCompleted( "agent_id": data.AgentID, "session_id": data.SessionID, }) - s.refreshContinuationSummary(ctx, run, data.AgentID) + s.refreshContinuationSummary(ctx, run, run.AgentProfileID) s.recordRunOutputSummary(ctx, run, *data) // run came from GetClaimedTasklessRunForAgent: it is the run that // actually launched. Taskless runs typically carry no task_id, so this @@ -430,27 +426,28 @@ func (s *Service) handleTasklessAgentCompleted( } // refreshContinuationSummary rebuilds the continuation summary for the -// given agent and upserts it under a scope keyed off the wakeup that -// produced the run. Errors are logged at warn — the prior row stays -// intact (last-good wins) and the run completion proceeds. +// given agent and upserts it under run.ContinuationScope — the scope key +// models.ContinuationScopeForRun computed once, at run-creation time, and +// persisted onto the row (see runs/repository/sqlite.CreateRun). Errors +// are logged at warn — the prior row stays intact (last-good wins) and +// the run completion proceeds. // -// Scope rules (office-heartbeat-as-routine): -// - run came from a routine wakeup (payload has routine_id) → -// "routine:" so each routine bridges its own context. -// - any other source (self / user / direct dispatch) → -// "agent:" so the agent still has somewhere to land its -// summary even when no routine is in play. -// -// The legacy "heartbeat" scope is retired alongside the agent-level -// heartbeat cron — every scheduled wake now flows through a routine. +// This deliberately reads the persisted field rather than recomputing it: +// run here comes from resolveLifecycleRun's fresh DB fetch, which can +// observe a context_snapshot a routine wakeup coalesced into this run +// after it was claimed (MarkWakeupRequestCoalesced patches only +// context_snapshot). Recomputing against that drifted snapshot could +// disagree with the scope the claim-time scheduler used when it read the +// prior continuation summary into the prompt — the persisted column is +// the single source of truth both sides read. func (s *Service) refreshContinuationSummary( - ctx context.Context, run *models.Run, agentID string, + ctx context.Context, run *models.Run, agentProfileID string, ) { - if s.repo == nil || run == nil || agentID == "" { + if s.repo == nil || run == nil || agentProfileID == "" { return } - scope := summaryScopeForRun(run, agentID) - inputs, err := summaryLoadInputs(ctx, s.repo, run, agentID, scope) + scope := run.ContinuationScope + inputs, err := summaryLoadInputs(ctx, s.repo, run, agentProfileID, scope) if err != nil { s.logger.Warn("continuation-summary load inputs failed", zap.String("run_id", run.ID), zap.Error(err)) @@ -458,7 +455,7 @@ func (s *Service) refreshContinuationSummary( } body := summaryBuild(inputs) upsertErr := s.repo.UpsertContinuationSummary(ctx, sqlite.AgentContinuationSummary{ - AgentProfileID: agentID, + AgentProfileID: agentProfileID, Scope: scope, Content: body, ContentTokens: approxTokenCount(body), @@ -477,37 +474,6 @@ func approxTokenCount(s string) int { return (len(s) + 3) / 4 } -// summaryScopeForRun returns the (agent_profile_id, scope) scope value -// for the continuation-summary upsert. Reads run.ContextSnapshot for a -// routine_id (set by the wakeup dispatcher when source="routine") and -// returns "routine:" when present; falls back to "agent:" so -// non-routine fires still have a stable upsert key. -func summaryScopeForRun(run *models.Run, agentID string) string { - if run == nil { - return "agent:" + agentID - } - if id := extractRoutineID(run.ContextSnapshot); id != "" { - return "routine:" + id - } - return "agent:" + agentID -} - -// extractRoutineID pulls routine_id out of a JSON snapshot. Returns "" -// for missing / malformed payloads so the caller falls back to the -// agent-scoped summary key. -func extractRoutineID(snapshot string) string { - if snapshot == "" { - return "" - } - var p struct { - RoutineID string `json:"routine_id"` - } - if err := json.Unmarshal([]byte(snapshot), &p); err != nil { - return "" - } - return p.RoutineID -} - // recordRunOutputSummary persists the agent's final message as the run's // output_summary. Best-effort — same contract as refreshContinuationSummary // above: any failure is logged at warn and never fails the completion diff --git a/apps/backend/internal/office/service/prompt_builder.go b/apps/backend/internal/office/service/prompt_builder.go index b7ab76ee04..8f045a3892 100644 --- a/apps/backend/internal/office/service/prompt_builder.go +++ b/apps/backend/internal/office/service/prompt_builder.go @@ -460,13 +460,10 @@ type BuildAgentPromptResult struct { // resume, only the wake context is included because the agent CLI // retains instructions from the previous session. // -// PR 1 of office-heartbeat-rework added taskless-run support: when -// taskID is empty AND continuationSummary is non-empty, the summary -// is prepended (sliced to 1,500 chars) before AGENTS.md so the agent -// has the "## Active focus" + "## Open blockers" + "## Next action" -// context without resuming a stale conversation. Today no caller -// passes taskID=="" so this branch is dead until PR 2 wires the -// agent_heartbeat cron. +// Taskless-run support prepends a non-empty continuation summary (sliced to +// 1,500 chars) before AGENTS.md. This gives routine and other taskless wakes +// the "## Active focus", "## Open blockers", and "## Next action" context +// without resuming a stale conversation. // // agentsMD is the AGENTS.md content read from the manifest (in-memory). // Sibling references like `./HEARTBEAT.md` have already been rewritten diff --git a/apps/backend/internal/office/service/scheduler_integration.go b/apps/backend/internal/office/service/scheduler_integration.go index 1cc5abad37..4c6032901e 100644 --- a/apps/backend/internal/office/service/scheduler_integration.go +++ b/apps/backend/internal/office/service/scheduler_integration.go @@ -382,7 +382,7 @@ func (si *SchedulerIntegration) assembleAgentPrompt( } } - continuationSummary := si.loadContinuationSummary(ctx, agent.ID, taskID) + continuationSummary := si.loadContinuationSummary(ctx, run, agent.ID, taskID) promptResult := si.svc.BuildAgentPrompt( run, agent, instructionsDir, agentsMD, isResume, wakeContext, taskID, continuationSummary, @@ -394,18 +394,26 @@ func (si *SchedulerIntegration) assembleAgentPrompt( // loadContinuationSummary fetches the per-(agent, scope) continuation // summary for a taskless run. Returns "" when: // - taskID is non-empty (task-bound runs don't use the summary doc) -// - the summary table has no row yet (first heartbeat ever for the agent) +// - the summary table has no row yet (first wake ever for the scope) // - the lookup errors out (best-effort, fall back to no-summary) // -// Today no caller passes taskID==""; this branch is a no-op until -// PR 2 lands the agent_heartbeat cron handler. +// The scope read here is run.ContinuationScope — the same key +// models.ContinuationScopeForRun computed once, at run-creation time, and +// that refreshContinuationSummary (event_subscribers.go) later writes +// under. This deliberately does NOT recompute the scope from run's +// ContextSnapshot: a routine wakeup that coalesces into this run after +// claim (MarkWakeupRequestCoalesced) patches only context_snapshot, so a +// second derivation here — against a run object that may itself predate +// that patch — could disagree with what the writer used at completion. +// Reading the persisted field closes that race for every caller instead +// of relying on ordering between two independent derivations. func (si *SchedulerIntegration) loadContinuationSummary( - ctx context.Context, agentID, taskID string, + ctx context.Context, run *models.Run, agentID, taskID string, ) string { - if taskID != "" || agentID == "" { + if run == nil || taskID != "" || agentID == "" { return "" } - prior, err := si.svc.repo.GetContinuationSummary(ctx, agentID, "heartbeat") + prior, err := si.svc.repo.GetContinuationSummary(ctx, agentID, run.ContinuationScope) if err != nil || prior == nil { return "" } diff --git a/apps/backend/internal/office/service/test_helpers.go b/apps/backend/internal/office/service/test_helpers.go index 1a741b8a50..f5b2237c76 100644 --- a/apps/backend/internal/office/service/test_helpers.go +++ b/apps/backend/internal/office/service/test_helpers.go @@ -2,6 +2,7 @@ package service import ( "context" + "fmt" "testing" "time" @@ -26,6 +27,15 @@ func (s *Service) GetContinuationSummaryForTest( return s.repo.GetContinuationSummary(ctx, agentProfileID, scope) } +// LoadContinuationSummaryForTest exposes SchedulerIntegration's private +// loadContinuationSummary so tests can drive the real reader path +// (as assembleAgentPrompt calls it) without duplicating its scope logic. +func (si *SchedulerIntegration) LoadContinuationSummaryForTest( + ctx context.Context, run *models.Run, agentID, taskID string, +) string { + return si.loadContinuationSummary(ctx, run, agentID, taskID) +} + // ListTasksTouchedByRunForTest exposes the repo's read query so the // run-lifecycle integration tests can verify run_id plumbing on the // activity log. @@ -59,6 +69,32 @@ func BuildPromptContextForTest(svc *Service, ctx context.Context, reason, payloa return si.buildPromptContext(ctx, reason, payload) } +// CoalesceRoutineWakeupForTest creates a wakeup-request carrying the +// given routine_id and coalesces it into an already-claimed run, exactly +// as the wakeup dispatcher does when a routine fire lands on an +// in-flight run (wakeup.Dispatcher.Dispatch -> MarkWakeupRequestCoalesced). +// Used by WO-16's continuation-scope regression tests to reproduce the +// claim-time vs completion-time snapshot drift without standing up the +// full wakeup dispatcher. +func (s *Service) CoalesceRoutineWakeupForTest( + ctx context.Context, t *testing.T, agentProfileID, runID, routineID string, +) { + t.Helper() + req := &sqlite.WakeupRequest{ + ID: "wakeup-" + runID + "-" + routineID, + AgentProfileID: agentProfileID, + Source: "routine", + Reason: "routine_trigger", + Payload: fmt.Sprintf(`{"routine_id":%q}`, routineID), + } + if err := s.repo.CreateWakeupRequest(ctx, req); err != nil { + t.Fatalf("create wakeup request: %v", err) + } + if err := s.repo.MarkWakeupRequestCoalesced(ctx, req.ID, runID); err != nil { + t.Fatalf("coalesce wakeup request into run: %v", err) + } +} + // ExecSQL executes raw SQL against the service's database for test setup. func (s *Service) ExecSQL(t *testing.T, query string, args ...interface{}) { t.Helper() diff --git a/apps/backend/internal/office/summary/inputs.go b/apps/backend/internal/office/summary/inputs.go index 8301a71a5a..64c4df5d01 100644 --- a/apps/backend/internal/office/summary/inputs.go +++ b/apps/backend/internal/office/summary/inputs.go @@ -25,7 +25,10 @@ type Repo interface { // LoadInputs assembles a BuildInputs from the database for a freshly // completed taskless run. Caller passes the run row (we read its // result_json + workspace via the agent profile join), the agent -// profile id, and the scope key (e.g. "heartbeat" or "routine:"). +// profile id, and the scope key persisted on the run: +// "routine:" for a routine-dispatched wake, or +// "agent:" otherwise. The legacy "heartbeat" scope is +// retired. // // Best-effort: any sub-query that fails is logged-by-virtue-of being // silently dropped, so a partial summary always wins over no summary. diff --git a/apps/backend/internal/runs/repository/sqlite/runs.go b/apps/backend/internal/runs/repository/sqlite/runs.go index 82e3fb64a8..5f99f20bf4 100644 --- a/apps/backend/internal/runs/repository/sqlite/runs.go +++ b/apps/backend/internal/runs/repository/sqlite/runs.go @@ -16,25 +16,33 @@ import ( ) // CreateRun creates a new run queue entry. +// +// ContinuationScope is decided here, once, before the row exists to be +// coalesced into — never at read or write time downstream. A routine +// wakeup that later coalesces into this run (MarkWakeupRequestCoalesced) +// only ever patches context_snapshot, so every later reader/writer of +// this run's continuation summary reads the value persisted here instead +// of re-deriving it against a snapshot that may have drifted. func (r *Repository) CreateRun(ctx context.Context, req *models.Run) error { if req.ID == "" { req.ID = uuid.New().String() } ensureRunDefaults(req) req.RequestedAt = time.Now().UTC() + req.ContinuationScope = models.ContinuationScopeForRun(req, req.AgentProfileID) _, err := r.db.ExecContext(ctx, r.db.Rebind(` INSERT INTO runs ( id, agent_profile_id, reason, payload, status, coalesced_count, idempotency_key, context_snapshot, capabilities, input_snapshot, output_summary, failure_reason, session_id, retry_count, scheduled_retry_at, - requested_at, error_message, cancel_reason - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + requested_at, error_message, cancel_reason, continuation_scope + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `), req.ID, req.AgentProfileID, req.Reason, req.Payload, req.Status, req.CoalescedCount, req.IdempotencyKey, req.ContextSnapshot, req.Capabilities, req.InputSnapshot, req.OutputSummary, req.FailureReason, req.SessionID, req.RetryCount, req.ScheduledRetryAt, req.RequestedAt, - req.ErrorMessage, req.CancelReason) + req.ErrorMessage, req.CancelReason, req.ContinuationScope) return err } diff --git a/apps/backend/internal/runs/repository/sqlite/runs_crud_test.go b/apps/backend/internal/runs/repository/sqlite/runs_crud_test.go index 7c1f026bb5..566e8f734c 100644 --- a/apps/backend/internal/runs/repository/sqlite/runs_crud_test.go +++ b/apps/backend/internal/runs/repository/sqlite/runs_crud_test.go @@ -61,6 +61,7 @@ func TestCreateRun_PersistsEveryWrittenColumn(t *testing.T) { checkString(t, "failure_reason", got.FailureReason, want.FailureReason) checkString(t, "session_id", got.SessionID, want.SessionID) checkString(t, "error_message", got.ErrorMessage, want.ErrorMessage) + checkString(t, "continuation_scope", got.ContinuationScope, "agent:agent-42") checkInt(t, "coalesced_count", got.CoalescedCount, want.CoalescedCount) checkInt(t, "retry_count", got.RetryCount, want.RetryCount) checkStringPtr(t, "idempotency_key", got.IdempotencyKey, want.IdempotencyKey) @@ -73,6 +74,19 @@ func TestCreateRun_PersistsEveryWrittenColumn(t *testing.T) { } } +func TestCreateRun_PersistsRoutineContinuationScope(t *testing.T) { + repo := newTestRepo(t) + want := &models.Run{ + AgentProfileID: "agent-routine", + Reason: "routine_trigger", + ContextSnapshot: `{"routine_id":"routine-1"}`, + } + mustCreateRun(t, repo, want) + + got := mustGetRun(t, repo, want.ID) + checkString(t, "continuation_scope", got.ContinuationScope, "routine:routine-1") +} + // TestCreateRun_LeavesUnwrittenColumnsAtTheirDefaults pins the other // half of the INSERT contract: the columns CreateRun deliberately does // not write (routing, prompt artifacts, timing) must land on their diff --git a/docs/specs/office/system-design/scheduler-02.md b/docs/specs/office/system-design/scheduler-02.md index 4880df2bca..762b4b9d86 100644 --- a/docs/specs/office/system-design/scheduler-02.md +++ b/docs/specs/office/system-design/scheduler-02.md @@ -53,7 +53,7 @@ INDEX (idempotency_key) ``` agent_profile_id TEXT NOT NULL -scope TEXT NOT NULL -- routine: +scope TEXT NOT NULL -- routine: or agent: content TEXT NOT NULL DEFAULT '' -- markdown, capped 8 KB content_tokens INT NOT NULL DEFAULT 0 updated_at TIMESTAMP NOT NULL @@ -116,8 +116,15 @@ started_at, completed_at result_json TEXT NOT NULL DEFAULT '{}' -- structured adapter output for summary builder assembled_prompt TEXT NOT NULL DEFAULT '' -- final prompt as the agent saw it (inspection) summary_injected TEXT NOT NULL DEFAULT '' -- the summary that was prepended (per-run snapshot) +continuation_scope TEXT NOT NULL DEFAULT '' -- summary key selected at run creation ``` +The run repository selects `continuation_scope` once when it creates a row: +`routine:` when the initial context has a routine ID, otherwise +`agent:`. Coalescing can change `context_snapshot`, but it +does not change the scope. The first wake that creates the run owns its +continuation-summary chain. + `runs.payload.task_id` is empty for taskless runs; all other run lifecycle fields (`idempotency_key`, claim/coalesce, cost rollup) carry over unchanged. ### Workspace settings (additions) @@ -188,7 +195,7 @@ type TaskAssignedPayload struct { TaskID string `json:"task_id"` } ### Run inspection (existing `/office/agents/[id]/runs/[runId]`) -`GetRunDetail` returns the `RunDetail` aggregate including the new columns (`result_json`, `assembled_prompt`, `summary_injected`) plus existing `context_snapshot`, `output_summary`. The detail UI surfaces a "Prompt" tab rendering the assembled prompt + injected summary. WS live updates via `useRunLiveSync` are unchanged. +`GetRunDetail` returns the `RunDetail` aggregate including the new columns (`result_json`, `assembled_prompt`, `summary_injected`, `continuation_scope`) plus existing `context_snapshot`, `output_summary`. The detail UI surfaces a "Prompt" tab rendering the assembled prompt + injected summary. WS live updates via `useRunLiveSync` are unchanged. ### Frontend UI @@ -282,7 +289,7 @@ A formal per-route authorization model (workspace membership, admin role, RBAC o ## Persistence guarantees -Survives a kandev process restart: all `agent_wakeup_requests` rows including `queued`, `claimed` (re-claimed on restart), and `scheduled_retry_at`; all `office_routines`, `office_routine_triggers` (with `next_run_at` advanced), `office_routine_runs` history; `agent_continuation_summaries` (last-good); `runs` rows including `result_json`, `assembled_prompt`, `summary_injected` snapshots. +Survives a kandev process restart: all `agent_wakeup_requests` rows including `queued`, `claimed` (re-claimed on restart), and `scheduled_retry_at`; all `office_routines`, `office_routine_triggers` (with `next_run_at` advanced), `office_routine_runs` history; `agent_continuation_summaries` (last-good); `runs` rows including `result_json`, `assembled_prompt`, `summary_injected`, and `continuation_scope` snapshots. Does NOT survive (reconstructed on next tick): in-memory claim leases - a `claimed` wakeup whose process died is picked up by the staleness/recovery path; the scheduler's claim query is the source of truth. The unstarted-task recovery sweep suppresses duplicates with its `NOT EXISTS` check on `runs`: any queued, claimed, or finished run of any age blocks redispatch, while failed and cancelled runs do not.