fix(office): stop agent continuation summaries from always reading back empty (#2971)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-08-24 19:16:26 +00:00
committed by GitHub
parent 727d59ce71
commit de3abefdb4
16 changed files with 617 additions and 95 deletions
@@ -0,0 +1,46 @@
package models
import "encoding/json"
// ContinuationScopeForRun computes the continuation-summary scope key for
// a run: "routine:<routine_id>" when the run's context snapshot carries a
// routine_id (set by the wakeup dispatcher for source="routine" wakeups),
// else "agent:<agent_profile_id>" 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
}
@@ -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)
}
})
}
}
+14 -4
View File
@@ -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
@@ -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
@@ -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
@@ -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)
}
}
@@ -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)
}
}
@@ -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:<id>"
// (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:<id> 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:<id>" instead of
// "agent:<id>". 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:<id> 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:<id>" 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:<id>") 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)
}
}
@@ -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:<routine_id>" so each routine bridges its own context.
// - any other source (self / user / direct dispatch) →
// "agent:<agent_id>" 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:<id>" when present; falls back to "agent:<id>" 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
@@ -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
@@ -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 ""
}
@@ -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()
@@ -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:<id>").
// profile id, and the scope key persisted on the run:
// "routine:<routine_id>" for a routine-dispatched wake, or
// "agent:<agent_profile_id>" 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.
@@ -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
}
@@ -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
@@ -53,7 +53,7 @@ INDEX (idempotency_key)
```
agent_profile_id TEXT NOT NULL
scope TEXT NOT NULL -- routine:<routine_id>
scope TEXT NOT NULL -- routine:<routine_id> or agent:<agent_profile_id>
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:<id>` when the initial context has a routine ID, otherwise
`agent:<agent_profile_id>`. 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.