From 1d7adaac230f8e8478bcec79d140c76ae4d805b1 Mon Sep 17 00:00:00 2001 From: Mobilyze Agents Date: Thu, 30 Jul 2026 08:37:53 +0000 Subject: [PATCH] fix: converge Desktop titles natively [closes BEAR-59] Stage deterministic title operations for the retained source task so native Codex mutations, rather than detached App Server writes, drive visible catalog convergence. Co-authored-by: open-swe[bot] --- CHANGELOG.md | 2 + INSTALL.md | 17 +- assets/AGENTS.threadbear.md | 37 ++ assets/embed_test.go | 30 +- assets/skill/SKILL.md | 39 +- cmd/threadbear/help.go | 16 +- cmd/threadbear/help_test.go | 6 +- cmd/threadbear/main.go | 21 +- cmd/threadbear/main_test.go | 19 + docs/architecture.md | 8 +- docs/compatibility.md | 2 +- docs/release-checklist.md | 15 +- internal/app/app.go | 79 ++-- internal/app/inspect.go | 14 + internal/app/status.go | 30 ++ internal/codex/index.go | 31 ++ internal/codex/index_test.go | 11 + internal/install/guide_test.go | 6 +- internal/output/result.go | 147 +++++- internal/state/model.go | 15 +- internal/titlebatch/titlebatch.go | 449 +++++++++++++++++++ internal/titlebatch/titlebatch_test.go | 219 +++++++++ internal/watch/cycle.go | 144 +++--- internal/watch/cycle_test.go | 598 +++++++++---------------- internal/watch/mutations.go | 24 +- scripts/replay-title-batch.mjs | 57 +++ site/install | 17 +- 27 files changed, 1507 insertions(+), 546 deletions(-) create mode 100644 internal/titlebatch/titlebatch.go create mode 100644 internal/titlebatch/titlebatch_test.go create mode 100644 scripts/replay-title-batch.mjs diff --git a/CHANGELOG.md b/CHANGELOG.md index 8d94af9..cab52a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ ### Fixed +- Restored mounted Codex Desktop title convergence through a deterministic retained-source native batch, with exact identity/revision/title guards, aggregate reports, canonical verification, and no child or extra semantic turn. + - Decoupled guided-install health from background historical convergence, restored context-sized semantic packing, added aggregate first-sweep progress, and isolated each heartbeat's classifier in one private minimal-auth App Server process reused across batches. - Confirmed the Codex 0.146 production isolation canary and four 200-observation rehearsals; ordinary status-guided work was fully deterministic, while serial remains the default because bounded first progress regressed. diff --git a/INSTALL.md b/INSTALL.md index 091ba12..f48fc4b 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -196,6 +196,12 @@ verifies the published checksum, candidate health, and embedded version before delegating to the candidate. The official-download promise belongs in the opening turn above; do not repeat it as a separate progress message. +## Native title convergence during guided install + +The person's installation consent enables only the managed title-batch protocol in the retained control task. It is not recurring consent for other native operations and must not be requested again on later control-task turns. After the bounded post-install handoff, stage the exact terminal footer, then run the fixed native title-batch program embedded in the installed ThreadBear AGENTS block inside one `functions.exec`. Drain every deterministic operation staged within the handoff window, but do not wait for unresolved semantic classification. Revalidate every operation immediately before the native setter and return only aggregate accepted, failed, drifted, and rejected operation IDs. + +A successful install close may claim visible title convergence only for accepted operations that ThreadBear subsequently verifies canonically. Missing native capability, failed mutation, identity disagreement, drift, or unverified canonical persistence is an explicit partial/failure result. Never expose title or manifest content to the model, create a child task, use the persistent task for unrelated work, drive the UI, restart Codex, or use private IPC/cache/database mutation. + ## 3. Identify this task backstage Before downloading, installing, renaming, pinning, or changing any managed resource, feature-detect the available Codex task tooling: @@ -816,10 +822,13 @@ outcomes below. When it is retryable, keep the healthy-install sentence, say plainly that ThreadBear will keep working on the aggregate retry count, and do not claim convergence. -The background heartbeat performs managed title writes directly through the -pinned App Server. Each title mutation is revision/title revalidated, journaled as applying and applied, verified through a fresh inventory read, journaled as -verified, and only then followed by any same-task archive. Do not run a separate -title worker, helper manifest, native report, child model, or replay step. +The background heartbeat deterministically stages exact, revision/title-guarded +title operations and never treats detached App Server persistence as Desktop +repaint proof. The retained, opted-in control task drains staged operations +through the fixed native title-batch program, reports only aggregate operation +IDs, and leaves canonical verification and commit to ThreadBear. A same-task +archive waits for accepted native application and canonical settlement. Do not +create a title child, recurring hosted automation, or control-task relay. Feature-detect and fail closed when required App Server methods or the tool-free classifier boundary are absent. The ordinary App Server stays on the real Codex diff --git a/assets/AGENTS.threadbear.md b/assets/AGENTS.threadbear.md index 6f64624..ca79d68 100644 --- a/assets/AGENTS.threadbear.md +++ b/assets/AGENTS.threadbear.md @@ -1,5 +1,42 @@ # ThreadBear +## Native title batch + +Guided-install consent enables this protocol only in the retained ThreadBear control task. Never ask again on later turns, and never use this consent for any other native operation. Other tasks use only the status footer. + +In the retained control task, after all substantive response work, choose the exact terminal ThreadBear footer. Stage that exact footer with `~/.local/bin/threadbear title-batch --json --stage >/dev/null`, sending one strict JSON object on stdin whose only field is `footer`, equal to the final line. Every nested `exec_command` process inherits the canonical public `CODEX_THREAD_ID`; the helper resolves and validates it against ThreadBear's persisted control-task identity without exposing the identifier to V8 or assistant text. The stage response contains no operation, title, or manifest data and must remain discarded. + +Immediately after staging, run one actual `functions.exec` containing only the fixed program below. Do not interpolate, add, remove, or revise code. The program obtains operation JSON inside the tool runtime, revalidates each operation immediately, calls only the native title capability, reports aggregate operation IDs, and never exposes titles, manifests, task contents, or helper output to assistant context. + +```js +const q=v=>"'"+String(v).replace(/'/g,"'\\''")+"'"; +const run=async cmd=>JSON.parse((await tools.exec_command({cmd})).output); +const valid=x=>x&&typeof x.operation_id==="string"&&typeof x.task_id==="string"&&typeof x.expected_revision==="string"&&typeof x.expected_title==="string"&&typeof x.desired_title==="string"&&x.desired_title!==""; +const batch=await run("~/.local/bin/threadbear title-batch --json --list"); +if(!batch||!Array.isArray(batch.plans))throw new Error("invalid_title_batch"); +const reports=[]; +for(const planned of batch.plans){ + if(!valid(planned)){throw new Error("invalid_title_operation");} + const guarded=await run("~/.local/bin/threadbear title-batch --json --operation "+q(planned.operation_id)); + if(!guarded||!Array.isArray(guarded.plans)||guarded.plans.length!==1){reports.push({operation_id:planned.operation_id,outcome:"drifted"});continue;} + const exact=guarded.plans[0]; + if(!valid(exact)||exact.operation_id!==planned.operation_id){reports.push({operation_id:planned.operation_id,outcome:"failed",error_code:"invalid_operation"});continue;} + try{await tools.codex_app__set_thread_title({threadId:exact.task_id,title:exact.desired_title});reports.push({operation_id:exact.operation_id,outcome:"accepted"});} + catch{reports.push({operation_id:exact.operation_id,outcome:"failed",error_code:"native_set_failed"});} +} +let result={accepted_ids:[],failed_ids:[],drifted_ids:[],rejected_ids:[]}; +let canonical=[]; +if(reports.length){ + const body=JSON.stringify({reports}); + result=await run("printf %s "+q(body)+" | ~/.local/bin/threadbear title-batch --json --report"); + const verified=await run("~/.local/bin/threadbear title-batch --json --list"); + canonical=verified.dispositions.filter(x=>x.outcome==="canonical_verified"||x.outcome==="canonical_verified_awaiting_footer").map(x=>x.operation_id); +} +text(JSON.stringify({accepted_ids:result.accepted_ids,canonical_ids:canonical,failed_ids:result.failed_ids,drifted_ids:result.drifted_ids,rejected_ids:result.rejected_ids})); +``` + +If staging, capability detection, native mutation, or reporting fails, do not claim visible convergence. Leave the operation pending and report the stable partial/failure outcome. After the V8 call, make no further tool call or commentary: send the substantive final response immediately and end it with the exact staged footer. If new input or state makes the operation stale, abort and recompute on the next retained control-task turn. + ## Status footer End each terminal response with exactly one compact status line. Use the matching literal example as its shape: diff --git a/assets/embed_test.go b/assets/embed_test.go index d13e99b..3a1d6ae 100644 --- a/assets/embed_test.go +++ b/assets/embed_test.go @@ -1,6 +1,7 @@ package assets import ( + "os/exec" "strings" "testing" ) @@ -48,12 +49,37 @@ func TestManagedSkillConversationalContract(t *testing.T) { } } -func TestManagedGuidanceContainsNoExecutableTitleActuator(t *testing.T) { +func TestManagedGuidanceBindsSourceNativeBatchOrdering(t *testing.T) { for name, content := range map[string]string{"agents": AgentsManagedContent, "skill": SkillManagedContent} { - for _, forbidden := range []string{"title-plan", "THREADBEAR_TITLE_ACTUATOR", "codex_app__create_thread", "codex_app__set_thread_title", "child actuator", "```js"} { + for _, required := range []string{"title-batch --json --stage", "title-batch --json --list", "title-batch --json --operation", "codex_app__set_thread_title", "title-batch --json --report", "no further tool call or commentary"} { + if !strings.Contains(content, required) { + t.Fatalf("%s missing native batch surface %q", name, required) + } + } + ordered := []string{"--list", "--operation", "codex_app__set_thread_title", "--report", "text(JSON.stringify"} + position := 0 + for _, token := range ordered { + next := strings.Index(content[position:], token) + if next < 0 { + t.Fatalf("%s missing ordered native batch token %q", name, token) + } + position += next + len(token) + } + for _, forbidden := range []string{"title-plan --json --batch", "THREADBEAR_TITLE_ACTUATOR", "codex_app__create_thread", "child actuator", "set_thread_archived", "process.env", "source_task_id"} { if strings.Contains(content, forbidden) { t.Fatalf("%s contains retired actuator surface %q", name, forbidden) } } } } + +func TestManagedTitleBatchRunsInFreshV8WithoutNodeGlobals(t *testing.T) { + command := exec.Command("node", "../scripts/replay-title-batch.mjs") + output, err := command.CombinedOutput() + if err != nil { + t.Fatalf("fresh V8 replay failed: %v\n%s", err, output) + } + if !strings.Contains(string(output), "fresh V8 title batch replay passed") { + t.Fatalf("unexpected replay output: %s", output) + } +} diff --git a/assets/skill/SKILL.md b/assets/skill/SKILL.md index 87e14d7..156f363 100644 --- a/assets/skill/SKILL.md +++ b/assets/skill/SKILL.md @@ -46,7 +46,44 @@ When explaining title state, use the canonical meanings: `⏳` running, `🚨` b ## Runtime roles -The persistent ThreadBear control task remains the user-facing master for help, configuration, install, update, uninstall, notices, user decisions, and exceptional recovery. Routine heartbeat classification uses fresh ephemeral App Server sessions and direct deterministic title/archive mutations; never route that work into the control task history. +### Native title batch + +Guided-install consent enables this protocol only in the retained ThreadBear control task. Never ask again on later turns, and never use this consent for any other native operation. Other tasks use only the status footer. + +In the retained control task, after all substantive response work, choose the exact terminal ThreadBear footer. Stage that exact footer with `~/.local/bin/threadbear title-batch --json --stage >/dev/null`, sending one strict JSON object on stdin whose only field is `footer`, equal to the final line. Every nested `exec_command` process inherits the canonical public `CODEX_THREAD_ID`; the helper resolves and validates it against ThreadBear's persisted control-task identity without exposing the identifier to V8 or assistant text. The stage response contains no operation, title, or manifest data and must remain discarded. + +Immediately after staging, run one actual `functions.exec` containing only the fixed program below. Do not interpolate, add, remove, or revise code. The program obtains operation JSON inside the tool runtime, revalidates each operation immediately, calls only the native title capability, reports aggregate operation IDs, and never exposes titles, manifests, task contents, or helper output to assistant context. + +```js +const q=v=>"'"+String(v).replace(/'/g,"'\\''")+"'"; +const run=async cmd=>JSON.parse((await tools.exec_command({cmd})).output); +const valid=x=>x&&typeof x.operation_id==="string"&&typeof x.task_id==="string"&&typeof x.expected_revision==="string"&&typeof x.expected_title==="string"&&typeof x.desired_title==="string"&&x.desired_title!==""; +const batch=await run("~/.local/bin/threadbear title-batch --json --list"); +if(!batch||!Array.isArray(batch.plans))throw new Error("invalid_title_batch"); +const reports=[]; +for(const planned of batch.plans){ + if(!valid(planned)){throw new Error("invalid_title_operation");} + const guarded=await run("~/.local/bin/threadbear title-batch --json --operation "+q(planned.operation_id)); + if(!guarded||!Array.isArray(guarded.plans)||guarded.plans.length!==1){reports.push({operation_id:planned.operation_id,outcome:"drifted"});continue;} + const exact=guarded.plans[0]; + if(!valid(exact)||exact.operation_id!==planned.operation_id){reports.push({operation_id:planned.operation_id,outcome:"failed",error_code:"invalid_operation"});continue;} + try{await tools.codex_app__set_thread_title({threadId:exact.task_id,title:exact.desired_title});reports.push({operation_id:exact.operation_id,outcome:"accepted"});} + catch{reports.push({operation_id:exact.operation_id,outcome:"failed",error_code:"native_set_failed"});} +} +let result={accepted_ids:[],failed_ids:[],drifted_ids:[],rejected_ids:[]}; +let canonical=[]; +if(reports.length){ + const body=JSON.stringify({reports}); + result=await run("printf %s "+q(body)+" | ~/.local/bin/threadbear title-batch --json --report"); + const verified=await run("~/.local/bin/threadbear title-batch --json --list"); + canonical=verified.dispositions.filter(x=>x.outcome==="canonical_verified"||x.outcome==="canonical_verified_awaiting_footer").map(x=>x.operation_id); +} +text(JSON.stringify({accepted_ids:result.accepted_ids,canonical_ids:canonical,failed_ids:result.failed_ids,drifted_ids:result.drifted_ids,rejected_ids:result.rejected_ids})); +``` + +If staging, capability detection, native mutation, or reporting fails, do not claim visible convergence. Leave the operation pending and report the stable partial/failure outcome. After the V8 call, make no further tool call or commentary: send the substantive final response immediately and end it with the exact staged footer. If new input or state makes the operation stale, abort and recompute on the next retained control-task turn. + +The persistent ThreadBear control task remains the user-facing master for help, configuration, install, update, uninstall, notices, user decisions, and exceptional recovery. Routine heartbeat classification uses fresh ephemeral App Server sessions. The heartbeat stages exact titles; only this retained opted-in task runs the fixed native title batch, while archive mutations remain deterministic. Never route routine classification or unrelated operations into the control task history. When managed global guidance is enabled, terminal responses use one concrete footer line: diff --git a/cmd/threadbear/help.go b/cmd/threadbear/help.go index 2849f57..ca975ed 100644 --- a/cmd/threadbear/help.go +++ b/cmd/threadbear/help.go @@ -120,6 +120,14 @@ func commandSpecFor(command app.Command) (commandSpec, bool) { flags.BoolVar(&request.TitlePlanDispatch, "dispatch", false, "return the retired fail-closed compatibility result") }}, true } + if command == app.CommandTitleBatch { + return commandSpec{command: command, synopsis: "Managed source-task native title protocol.", registerFlags: func(flags *flag.FlagSet, request *app.Request) { + flags.BoolVar(&request.TitleBatchList, "list", false, "return eligible exact title operations") + flags.BoolVar(&request.TitleBatchStage, "stage", false, "stage the source terminal footer from stdin") + flags.BoolVar(&request.TitleBatchReport, "report", false, "accept aggregate native results from stdin") + flags.StringVar(&request.TitleBatchOperation, "operation", "", "revalidate one operation ID") + }}, true + } for _, spec := range commandSpecs { if spec.command == command { return spec, true @@ -137,8 +145,8 @@ func newCommandFlagSet(spec commandSpec, request *app.Request) *flag.FlagSet { } func requestedHelp(args []string) (string, int, bool) { - if len(args) > 0 && app.Command(args[0]) == app.CommandTitlePlan && (containsFlag(args[1:], "-h") || containsFlag(args[1:], "--help")) { - return unknownCommandMessage(app.CommandTitlePlan), 2, true + if len(args) > 0 && (app.Command(args[0]) == app.CommandTitlePlan || app.Command(args[0]) == app.CommandTitleBatch) && (containsFlag(args[1:], "-h") || containsFlag(args[1:], "--help")) { + return unknownCommandMessage(app.Command(args[0])), 2, true } if len(args) == 0 { return renderTopLevelHelp(), 2, true @@ -151,8 +159,8 @@ func requestedHelp(args []string) (string, int, bool) { return renderTopLevelHelp(), 0, true } if len(args) == 2 { - if app.Command(args[1]) == app.CommandTitlePlan { - return unknownCommandMessage(app.CommandTitlePlan), 2, true + if app.Command(args[1]) == app.CommandTitlePlan || app.Command(args[1]) == app.CommandTitleBatch { + return unknownCommandMessage(app.Command(args[1])), 2, true } if spec, ok := commandSpecFor(app.Command(args[1])); ok { return renderCommandHelp(spec), 0, true diff --git a/cmd/threadbear/help_test.go b/cmd/threadbear/help_test.go index 4a55f93..748e4cf 100644 --- a/cmd/threadbear/help_test.go +++ b/cmd/threadbear/help_test.go @@ -16,8 +16,8 @@ var updateHelpGolden = flag.Bool("update", false, "update help golden files") func TestHelpCommandMetadataComplete(t *testing.T) { commands := app.AllCommands() - if len(commandSpecs) != len(commands)-1 { - t.Fatalf("commandSpecs has %d entries; want %d public commands", len(commandSpecs), len(commands)-1) + if len(commandSpecs) != len(commands)-2 { + t.Fatalf("commandSpecs has %d entries; want %d public commands", len(commandSpecs), len(commands)-2) } seen := make(map[app.Command]bool, len(commandSpecs)) for _, spec := range commandSpecs { @@ -33,7 +33,7 @@ func TestHelpCommandMetadataComplete(t *testing.T) { } } for _, command := range commands { - if command == app.CommandTitlePlan { + if command == app.CommandTitlePlan || command == app.CommandTitleBatch { continue } if !command.Valid() { diff --git a/cmd/threadbear/main.go b/cmd/threadbear/main.go index 4204a8d..d519968 100644 --- a/cmd/threadbear/main.go +++ b/cmd/threadbear/main.go @@ -28,6 +28,7 @@ import ( "github.com/ericlitman/threadbear/internal/output" "github.com/ericlitman/threadbear/internal/state" statusresolver "github.com/ericlitman/threadbear/internal/status" + "github.com/ericlitman/threadbear/internal/titlebatch" "github.com/ericlitman/threadbear/internal/titleplan" "github.com/ericlitman/threadbear/internal/tokens" updatepkg "github.com/ericlitman/threadbear/internal/update" @@ -207,8 +208,9 @@ func newOperatorService(installedVersion string, stdout, stderr io.Writer, forma return uninstaller, prompter.Close, nil } titlePlanCompatibility := titleplan.Service{} + titleBatch := titlebatch.Service{Store: store, Inventory: inventory, Input: os.Stdin, Now: time.Now, SourceIdentity: func() string { return os.Getenv("CODEX_THREAD_ID") }} service := app.NewWithOperatorCommands(installedVersion, app.OperatorDependencies{ - Store: store, Inventory: inventory, Clock: clock, LaunchAgent: launch, TitlePlanCompatibility: titlePlanCompatibility, + Store: store, Inventory: inventory, Clock: clock, LaunchAgent: launch, TitlePlanCompatibility: titlePlanCompatibility, TitleBatch: titleBatch, ManagedAgents: managed, Unarchiver: appServerUnarchiver{runtime: appServers}, Heartbeat: runner, Preview: func(preview output.PreviewResult) error { if request.NonInteractive { @@ -279,6 +281,23 @@ func (i *lazyInventory) Inventory(ctx context.Context, controlTaskID string) (co return i.index.Inventory(ctx, controlTaskID) } +func (i *lazyInventory) Task(ctx context.Context, taskID string) (codex.Task, error) { + i.mu.Lock() + defer i.mu.Unlock() + if i.index == nil { + sqliteHome, err := codex.ResolveSQLiteHome(i.codexHome) + if err != nil { + return codex.Task{}, err + } + index, err := codex.OpenIndex(sqliteHome) + if err != nil { + return codex.Task{}, err + } + i.index = index + } + return i.index.Task(ctx, taskID) +} + func (i *lazyInventory) Close() error { i.mu.Lock() defer i.mu.Unlock() diff --git a/cmd/threadbear/main_test.go b/cmd/threadbear/main_test.go index 952cb31..f1a656d 100644 --- a/cmd/threadbear/main_test.go +++ b/cmd/threadbear/main_test.go @@ -747,3 +747,22 @@ func TestRetiredTitlePlanCompatibilityIsHiddenAndFailClosed(t *testing.T) { t.Fatalf("got %q want %q", got, want) } } + +func TestManagedTitleBatchCommandIsHiddenAndStrict(t *testing.T) { + request, err := parseRequest([]string{"title-batch", "--json", "--list"}) + if err != nil || !request.TitleBatchList { + t.Fatalf("request=%+v err=%v", request, err) + } + operation, err := parseRequest([]string{"title-batch", "--json", "--operation", "op"}) + if err != nil || operation.TitleBatchOperation != "op" { + t.Fatalf("operation=%+v err=%v", operation, err) + } + for _, args := range [][]string{{"title-batch", "--list"}, {"title-batch", "--json"}, {"title-batch", "--json", "--list", "--report"}} { + if _, err := parseRequest(args); err == nil { + t.Fatalf("accepted %v", args) + } + } + if strings.Contains(renderTopLevelHelp(), "title-batch") { + t.Fatal("managed title batch command is visible in help") + } +} diff --git a/docs/architecture.md b/docs/architecture.md index a58054a..d1aeb6c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -7,8 +7,8 @@ ThreadBear is a single pure-Go macOS binary run by a user LaunchAgent. Its desig - **Heartbeat runner:** inventories every managed unarchived Codex Desktop task, excluding the control task. - **Deterministic resolver:** applies runtime, structured-error, automation, interruption, and valid footer evidence in precedence order. - **Ephemeral classifier:** sends only unresolved changed tasks to one isolated App Server per heartbeat using the configured model and effort (Luna medium by default). Its private temporary home contains only a regular `auth.json` copy plus files Codex creates there; every capacity-sized batch shares that process, and sessions never append to the control task. -- **Mutation layer:** performs revision/title-guarded App Server title writes and safe archives through the same crash-recoverable checkpoint journal. -- **State store:** atomically records the last completed snapshot, task classifications, archive records, retries, update checks, and delivered notice versions. Schema-v2 pending title plans are decoded only for one-time migration into ordinary checkpoint operations. +- **Mutation layer:** stages revision/title-guarded title operations for the opted-in retained source task, accepts aggregate native outcomes, verifies canonical persistence separately, and performs safe archives through the same crash-recoverable journal. +- **State store:** atomically records the last completed snapshot, task classifications, pending native title operations, canonical-verification outcomes, archive records, retries, update checks, and delivered notice versions. - **Control task:** one persistent task titled `🧵🐻 ThreadBear 🐻🧵` used for actionable notices, not classifier history. ## Heartbeat flow @@ -18,7 +18,7 @@ ThreadBear is a single pure-Go macOS binary run by a user LaunchAgent. Its desig 3. If nothing changed and no update check or version-drift work is due, exit without starting App Server, invoking a classifier, mutating state, or writing output. 4. Resolve changed tasks from deterministic evidence and read only the new rollout tail needed for the cumulative output-token figure. Mechanically settled tasks never reach Luna. 5. After deterministic resolution, create one private minimal-auth classifier App Server only when unresolved work remains. Pack latest turns into context-safe ephemeral calls on that shared process; previous evidence is read through the ordinary real-home App Server only for tasks that request it. -6. Revalidate title and archive operations before direct mutation. A title is journaled applying, written with `thread/name/set`, journaled applied, verified by inventory, journaled verified, and only then may the same task be archived. +6. Revalidate title operations, persist them in the native outbox, and commit classification without using detached `thread/name/set`. The retained control task revalidates each operation immediately before its native setter, reports aggregate operation IDs, and ThreadBear verifies canonical persistence before same-task archive. 7. Commit successful siblings and the captured snapshot atomically. Failed operations retain conservative state and bounded retry metadata. 8. When due, compare release metadata. With auto-update enabled, install a newer release through the verified replacement path; with it disabled, retain the once-daily notice-only behavior. After any version change, the new binary reconciles managed guidance and stages one changelog-backed control-task announcement. @@ -34,7 +34,7 @@ EMOJI durable subject → concise next action The action is omitted when none is warranted. The optional token display uses cumulative output tokens from the last rollout `token_count` event and renders a two-significant-figure magnitude at the start or labeled end of the managed title. ThreadBear caches the rollout path and last-read offset/size, so an unchanged rollout is not read again. -ThreadBear strips existing canonical status prefixes, preserves user-edited subjects, and records its last applied title plus the exact token segment it owns. It can update or remove its prefix, token figure, and action without consuming user ownership. Verified App Server persistence updates exact ThreadBear ownership in the committed snapshot. Desktop rendering remains outside the persistence contract and is never manipulated through private caches or UI automation. +ThreadBear strips existing canonical status prefixes, preserves user-edited subjects, and records its last applied title plus the exact token segment it owns. It can update or remove its prefix, token figure, and action without consuming user ownership. A native-success report and canonical inventory verification are separate required boundaries before ThreadBear commits exact title ownership. Rendered convergence uses only the capability-detected host-native setter in the retained control task; ThreadBear never edits private caches or drives the UI. Only `complete` tasks can be auto-archived. Running, blocked, needs-input, automation, next-steps, and unknown tasks remain active regardless of age. A manual unarchive or `~/.local/bin/threadbear restore TASK_ID` starts a new inactivity grace period. diff --git a/docs/compatibility.md b/docs/compatibility.md index b76afc0..15c82c8 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -18,7 +18,7 @@ ThreadBear depends on compatibility-detectable local Codex capabilities for comp ## Sidebar expectations -The heartbeat writes titles through capability-detected `thread/name/set`, then verifies persistence through a fresh inventory read before committing ownership or archiving the same task. ThreadBear does not promise private Desktop cache invalidation and never edits caches, uses private IPC/UI automation/daemons, or requires a restart. +The heartbeat stages deterministic title operations and does not call detached `thread/name/set` for routine title work. With explicit guided-install consent, only the retained control task may drain those operations through the capability-detected host-native title setter; ThreadBear accepts aggregate native outcomes and verifies canonical persistence separately before committing ownership or archiving the same task. Missing native capability remains an explicit pending convergence state. ThreadBear never edits caches, uses private IPC/UI automation/daemons, creates a title child, or requires a restart. ## LaunchAgent timing and environment diff --git a/docs/release-checklist.md b/docs/release-checklist.md index 602630b..38d5e08 100644 --- a/docs/release-checklist.md +++ b/docs/release-checklist.md @@ -33,15 +33,16 @@ Complete this checklist for every stable ThreadBear release. The hosted smoke explicitly does not prove real Codex auth, real App Server side effects, classification heartbeat behavior, title/archive effects, or an architecture other than its runner. The required replica rehearsal covers the Codex-touching composition before tagging. -## Direct title-write canaries +## Native title-convergence canaries Before a release that changes title handling, use a real Codex catalog and verify: -1. A changed task is revalidated before `thread/name/set`, and checkpoint saves record applying, applied, and verified in order. -2. Inventory persistence is verified before committed title ownership changes and before a same-task archive begins. -3. Applying/applied crash recovery does not repeat a title already visible in inventory and safely retries one not yet visible. -4. Schema-v2 pending plans drain once; valid plans use no evidence, transcript, token, or classifier reads, while missing and drifted plans return to ordinary comparison. -5. A same-title migrated refresh performs exactly one setter call, and a failed direct write retains the compatibility plan for retry. -6. One unchanged heartbeat performs zero classifier turns and zero title RPCs. +1. Detached `thread/name/set` canonical persistence is recorded separately and is not claimed as mounted Desktop convergence. +2. The retained control task resolves its canonical runtime identity exactly, drains a multi-title native batch, and repaints mounted Recents and saved-project rows without reload or restart. +3. Every operation is revalidated immediately before the native setter; unowned, malformed, stale, and drifted rows are skipped and reported only by operation ID. +4. Native success, canonical verification, checkpoint recovery, partial failure, duplicate reports, and bounded canonical retries settle deterministically before same-task archive. +5. The current source title matches the exact terminal ThreadBear footer, and a changed/missing final footer recomputes rather than accepting the staged status. +6. One unchanged heartbeat and retained source turn perform zero classifier turns and zero title operations. +7. A disposable 50+ task replica exercises the fixed aggregate raw-V8 path and retains only aggregate counts. Do not use private IPC, caches, daemons, UI automation, restarts, or the persistent ThreadBear control task for routine title work. diff --git a/internal/app/app.go b/internal/app/app.go index d05a0bc..8867fe2 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -14,25 +14,27 @@ import ( type Command string const ( - CommandInstall Command = "install" - CommandHeartbeat Command = "heartbeat" - CommandTitlePlan Command = "title-plan" - CommandStatus Command = "status" - CommandInspect Command = "inspect" - CommandConfigure Command = "configure" - CommandEnable Command = "enable" - CommandDisable Command = "disable" - CommandRestore Command = "restore" - CommandSelfTest Command = "self-test" - CommandUpdate Command = "update" - CommandUninstall Command = "uninstall" - CommandVersion Command = "version" + CommandInstall Command = "install" + CommandHeartbeat Command = "heartbeat" + CommandTitlePlan Command = "title-plan" + CommandTitleBatch Command = "title-batch" + CommandStatus Command = "status" + CommandInspect Command = "inspect" + CommandConfigure Command = "configure" + CommandEnable Command = "enable" + CommandDisable Command = "disable" + CommandRestore Command = "restore" + CommandSelfTest Command = "self-test" + CommandUpdate Command = "update" + CommandUninstall Command = "uninstall" + CommandVersion Command = "version" ) var allCommands = [...]Command{ CommandInstall, CommandHeartbeat, CommandTitlePlan, + CommandTitleBatch, CommandStatus, CommandInspect, CommandConfigure, @@ -75,18 +77,22 @@ func (p ConfigPatch) Empty() bool { } type Request struct { - Command Command - JSON bool - DryRun bool - Confirm bool - NonInteractive bool - Candidate bool - ArchiveControlTask bool - Version string - TaskID string - TitlePlanDispatch bool - ControlTaskID string - Configure ConfigPatch + Command Command + JSON bool + DryRun bool + Confirm bool + NonInteractive bool + Candidate bool + ArchiveControlTask bool + Version string + TaskID string + TitlePlanDispatch bool + TitleBatchList bool + TitleBatchStage bool + TitleBatchReport bool + TitleBatchOperation string + ControlTaskID string + Configure ConfigPatch } type Handler func(context.Context, Request) (output.Result, error) @@ -159,6 +165,29 @@ func (r Request) Validate() error { } else if r.TitlePlanDispatch { return fmt.Errorf("%w: title-plan flags are title-plan-only", ErrInvalidRequest) } + if strings.TrimSpace(r.TitleBatchOperation) != r.TitleBatchOperation { + return fmt.Errorf("%w: title-batch identity must not contain surrounding whitespace", ErrInvalidRequest) + } + if r.Command == CommandTitleBatch { + modes := 0 + if r.TitleBatchList { + modes++ + } + if r.TitleBatchStage { + modes++ + } + if r.TitleBatchReport { + modes++ + } + if r.TitleBatchOperation != "" { + modes++ + } + if !r.JSON || modes != 1 { + return fmt.Errorf("%w: title-batch requires --json and one strict mode", ErrInvalidRequest) + } + } else if r.TitleBatchList || r.TitleBatchStage || r.TitleBatchReport || r.TitleBatchOperation != "" { + return fmt.Errorf("%w: title-batch flags are title-batch-only", ErrInvalidRequest) + } if strings.TrimSpace(r.ControlTaskID) != r.ControlTaskID { return fmt.Errorf("%w: control task ID must not contain surrounding whitespace", ErrInvalidRequest) } diff --git a/internal/app/inspect.go b/internal/app/inspect.go index a4858c5..9096c74 100644 --- a/internal/app/inspect.go +++ b/internal/app/inspect.go @@ -86,6 +86,17 @@ func InspectHandler(store OperatorStore, inventory OperatorInventory, clock Oper retry = &output.RetryResult{TaskID: request.TaskID, Operation: diagnostic.Operation, ErrorCode: diagnostic.ErrorCode} } } + plan, pendingTitle := committed.PendingTitlePlans[request.TaskID] + renderedConvergence := "" + if pendingTitle { + renderedConvergence = "pending_native" + if plan.NativeOutcome == state.NativeTitleSucceeded { + renderedConvergence = "native_reported_pending_canonical" + } + if plan.NativeOutcome == state.NativeTitleFailed { + renderedConvergence = plan.NativeErrorCode + } + } eligible := recordMatchesCurrent && archiveEligibleForInspect(record, clock.Now().UTC(), cfg.ArchiveAfterDays) && cfg.ArchiveEnabled return output.InspectResult{ TaskID: request.TaskID, @@ -99,6 +110,9 @@ func InspectHandler(store OperatorStore, inventory OperatorInventory, clock Oper ManagedTokenPosition: managedTokenPosition, ManagedTokenDisplay: managedTokenDisplay, TokenUsageFound: tokenUsageFound, + PendingTitlePlan: pendingTitle, + NativeTitleOutcome: plan.NativeOutcome, + RenderedConvergence: renderedConvergence, }, nil } } diff --git a/internal/app/status.go b/internal/app/status.go index 1fa1fa8..a72213e 100644 --- a/internal/app/status.go +++ b/internal/app/status.go @@ -49,6 +49,13 @@ type TitlePlanCompatibility interface { Dispatch() output.Result } +type TitleBatch interface { + List(context.Context) (output.Result, error) + Operation(context.Context, string) (output.Result, error) + Stage(context.Context) (output.Result, error) + Report() (output.Result, error) +} + type OperatorDependencies struct { Store OperatorStore Inventory OperatorInventory @@ -60,6 +67,7 @@ type OperatorDependencies struct { Unarchiver Unarchiver Heartbeat HeartbeatRunner TitlePlanCompatibility TitlePlanCompatibility + TitleBatch TitleBatch Install Handler SelfTest Handler Update Updater @@ -74,6 +82,20 @@ func NewWithOperatorCommands(version string, deps OperatorDependencies) *Service return deps.TitlePlanCompatibility.Dispatch(), nil } } + if deps.TitleBatch != nil { + service.handlers[CommandTitleBatch] = func(ctx context.Context, request Request) (output.Result, error) { + switch { + case request.TitleBatchList: + return deps.TitleBatch.List(ctx) + case request.TitleBatchOperation != "": + return deps.TitleBatch.Operation(ctx, request.TitleBatchOperation) + case request.TitleBatchStage: + return deps.TitleBatch.Stage(ctx) + default: + return deps.TitleBatch.Report() + } + } + } service.handlers[CommandStatus] = StatusHandler(version, deps.Store, deps.LaunchAgent) service.handlers[CommandInspect] = InspectHandler(deps.Store, deps.Inventory, deps.Clock) service.handlers[CommandConfigure] = ConfigureHandler(deps.Store, deps.LaunchAgent, deps.Preview, deps.Confirm, deps.ManagedAgents) @@ -145,6 +167,12 @@ func StatusHandler(version string, store OperatorStore, launchAgent LaunchAgent) } else { firstSweep = committed.LastSweep } + nativeSuccesses := 0 + for _, plan := range committed.PendingTitlePlans { + if plan.NativeOutcome == state.NativeTitleSucceeded { + nativeSuccesses++ + } + } return output.StatusResult{ InstalledVersion: version, LaunchAgentHealthy: healthy, @@ -164,6 +192,8 @@ func StatusHandler(version string, store OperatorStore, launchAgent LaunchAgent) ClassifierContextBudgetBytes: cfg.ClassifierContextBudgetBytes, }, PendingRetries: len(pendingTaskIDs), + PendingTitlePlans: len(committed.PendingTitlePlans), + NativeTitleSuccesses: nativeSuccesses, LastUpdateCheck: committed.LastUpdateCheck, LastUpdateFailure: committed.LastUpdateFailure, LastReconcileFailure: committed.LastReconcileFailure, diff --git a/internal/codex/index.go b/internal/codex/index.go index b4539e2..712ff04 100644 --- a/internal/codex/index.go +++ b/internal/codex/index.go @@ -141,6 +141,37 @@ func (i *Index) Close() error { return i.db.Close() } +func (i *Index) Task(ctx context.Context, taskID string) (Task, error) { + if taskID == "" { + return Task{}, errors.New("task ID is required") + } + if err := i.validateSchema(ctx); err != nil { + return Task{}, err + } + var task Task + var updatedAtMS int64 + var name sql.NullString + var archived int + var threadSource sql.NullString + var rolloutPath sql.NullString + err := i.db.QueryRowContext(ctx, ` +SELECT id, updated_at_ms, title, name, archived, source, thread_source, rollout_path +FROM threads +WHERE id = ? AND source = 'vscode'`, taskID).Scan(&task.TaskID, &updatedAtMS, &task.DerivedTitle, &name, &archived, &task.Source, &threadSource, &rolloutPath) + if err != nil { + return Task{}, fmt.Errorf("read Codex task: %w", err) + } + task.Revision = strconv.FormatInt(updatedAtMS, 10) + task.Title = task.DerivedTitle + if name.Valid { + task.Title = name.String + } + task.Archived = archived != 0 + task.ThreadSource = threadSource.String + task.RolloutPath = rolloutPath.String + return task, nil +} + func (i *Index) Inventory(ctx context.Context, controlTaskID string) (Inventory, error) { if controlTaskID == "" { return Inventory{}, errors.New("control task ID is required") diff --git a/internal/codex/index_test.go b/internal/codex/index_test.go index 84f4021..2cee3b6 100644 --- a/internal/codex/index_test.go +++ b/internal/codex/index_test.go @@ -630,3 +630,14 @@ func directoryNames(t *testing.T, path string) []string { slices.Sort(names) return names } + +func TestTaskReadsExcludedControlByCanonicalID(t *testing.T) { + index := openFixture(t, "state_5.sqlite") + task, err := index.Task(context.Background(), "control-123") + if err != nil { + t.Fatal(err) + } + if task.TaskID != "control-123" || task.Source != "vscode" || task.Archived { + t.Fatalf("task=%+v", task) + } +} diff --git a/internal/install/guide_test.go b/internal/install/guide_test.go index d24555b..d8a6ac9 100644 --- a/internal/install/guide_test.go +++ b/internal/install/guide_test.go @@ -35,7 +35,7 @@ func TestCodexInstallGuideCarriesConversationContract(t *testing.T) { } } -func TestCodexInstallGuideUsesDirectAppServerTitleWrites(t *testing.T) { +func TestCodexInstallGuideUsesManagedSourceNativeTitleBatch(t *testing.T) { guide := readInstallGuide(t) published, err := os.ReadFile("../../site/install") if err != nil { @@ -44,9 +44,9 @@ func TestCodexInstallGuideUsesDirectAppServerTitleWrites(t *testing.T) { if guide != string(published) { t.Fatal("INSTALL.md and site/install differ") } - for _, required := range []string{"persistent title, archive, and unarchive methods", "journaled as applying and applied", "verified through a fresh inventory read", "only then followed by any same-task archive", "persistent control task remains reserved"} { + for _, required := range []string{"Native title convergence during guided install", "managed title-batch protocol", "Revalidate every operation immediately before the native setter", "accepted, failed, drifted, and rejected operation IDs", "do not wait for unresolved semantic classification"} { if !strings.Contains(guide, required) { - t.Fatalf("INSTALL.md missing direct title contract %q", required) + t.Fatalf("INSTALL.md missing source-native title contract %q", required) } } for _, forbidden := range []string{"title-plan --json --batch", "title-plan --json --report", "Child actuator phase", "codex_app__create_thread"} { diff --git a/internal/output/result.go b/internal/output/result.go index db20fd0..b17383c 100644 --- a/internal/output/result.go +++ b/internal/output/result.go @@ -42,6 +42,86 @@ func (r TitleDispatchResult) Human() string { return string(data) } +type TitleBatchItem struct { + OperationID string `json:"operation_id"` + TaskID string `json:"task_id"` + ExpectedRevision string `json:"expected_revision"` + ExpectedTitle string `json:"expected_title"` + DesiredTitle string `json:"desired_title"` +} + +type TitleBatchDisposition struct { + OperationID string `json:"operation_id,omitempty"` + TaskID string `json:"task_id,omitempty"` + Outcome string `json:"outcome"` +} + +type TitleBatchResult struct { + Version int `json:"version"` + Mode string `json:"mode"` + Plans []TitleBatchItem `json:"plans"` + Dispositions []TitleBatchDisposition `json:"dispositions"` +} + +func (TitleBatchResult) result() {} +func (TitleBatchResult) Empty() bool { return false } +func (r TitleBatchResult) Human() string { + data, _ := json.Marshal(r.normalized()) + return string(data) +} +func (r TitleBatchResult) normalized() TitleBatchResult { + if r.Version == 0 { + r.Version = CurrentResultVersion + } + r.Plans = slices.Clone(r.Plans) + r.Dispositions = slices.Clone(r.Dispositions) + sort.Slice(r.Plans, func(i, j int) bool { return r.Plans[i].OperationID < r.Plans[j].OperationID }) + sort.Slice(r.Dispositions, func(i, j int) bool { return r.Dispositions[i].OperationID < r.Dispositions[j].OperationID }) + if r.Plans == nil { + r.Plans = []TitleBatchItem{} + } + if r.Dispositions == nil { + r.Dispositions = []TitleBatchDisposition{} + } + return r +} + +type TitleBatchReportResult struct { + Version int `json:"version"` + AcceptedIDs []string `json:"accepted_ids"` + FailedIDs []string `json:"failed_ids"` + DriftedIDs []string `json:"drifted_ids"` + RejectedIDs []string `json:"rejected_ids"` +} + +func (TitleBatchReportResult) result() {} +func (TitleBatchReportResult) Empty() bool { return false } +func (r TitleBatchReportResult) Human() string { + data, _ := json.Marshal(r.normalized()) + return string(data) +} +func (r TitleBatchReportResult) normalized() TitleBatchReportResult { + if r.Version == 0 { + r.Version = CurrentResultVersion + } + for _, values := range [][]string{r.AcceptedIDs, r.FailedIDs, r.DriftedIDs, r.RejectedIDs} { + slices.Sort(values) + } + if r.AcceptedIDs == nil { + r.AcceptedIDs = []string{} + } + if r.FailedIDs == nil { + r.FailedIDs = []string{} + } + if r.DriftedIDs == nil { + r.DriftedIDs = []string{} + } + if r.RejectedIDs == nil { + r.RejectedIDs = []string{} + } + return r +} + type TaskChange struct { TaskID string `json:"task_id"` State state.TaskStatus `json:"state"` @@ -137,6 +217,8 @@ type StatusResult struct { ControlTaskID string `json:"control_task_id"` Preferences Preferences `json:"preferences"` PendingRetries int `json:"pending_retries"` + PendingTitlePlans int `json:"pending_title_plans"` + NativeTitleSuccesses int `json:"native_title_successes"` LastUpdateCheck *time.Time `json:"last_update_check,omitempty"` LastUpdateFailure *state.Failure `json:"last_update_failure,omitempty"` LastReconcileFailure *state.Failure `json:"last_reconcile_failure,omitempty"` @@ -156,7 +238,7 @@ func (r StatusResult) Human() string { if health == "unavailable" { health = "scheduler adapter unavailable (pending install unit)" } - message := fmt.Sprintf("ThreadBear %s · LaunchAgent %s · heartbeat %s · control task %s · heartbeat interval %ds · archive %t/%dd · rename %t · auto-update %t · token display %s · AGENTS %t · classifier %s/%s/%dB · retries %d · update check %s · update failure %s · reconcile failure %s", r.InstalledVersion, health, formatTime(r.LastCompletedHeartbeat), r.ControlTaskID, r.Preferences.HeartbeatSeconds, r.Preferences.ArchiveEnabled, r.Preferences.ArchiveAfterDays, r.Preferences.RenameEnabled, r.Preferences.AutoUpdateEnabled, r.Preferences.TokenDisplay, r.Preferences.AgentsEnabled, r.Preferences.ClassifierModel, r.Preferences.ClassifierEffort, r.Preferences.ClassifierContextBudgetBytes, r.PendingRetries, formatTime(r.LastUpdateCheck), formatFailure(r.LastUpdateFailure), formatFailure(r.LastReconcileFailure)) + message := fmt.Sprintf("ThreadBear %s · LaunchAgent %s · heartbeat %s · control task %s · heartbeat interval %ds · archive %t/%dd · rename %t · auto-update %t · token display %s · AGENTS %t · classifier %s/%s/%dB · retries %d · title plans %d (%d native pending canonical) · update check %s · update failure %s · reconcile failure %s", r.InstalledVersion, health, formatTime(r.LastCompletedHeartbeat), r.ControlTaskID, r.Preferences.HeartbeatSeconds, r.Preferences.ArchiveEnabled, r.Preferences.ArchiveAfterDays, r.Preferences.RenameEnabled, r.Preferences.AutoUpdateEnabled, r.Preferences.TokenDisplay, r.Preferences.AgentsEnabled, r.Preferences.ClassifierModel, r.Preferences.ClassifierEffort, r.Preferences.ClassifierContextBudgetBytes, r.PendingRetries, r.PendingTitlePlans, r.NativeTitleSuccesses, formatTime(r.LastUpdateCheck), formatFailure(r.LastUpdateFailure), formatFailure(r.LastReconcileFailure)) if r.FirstSweep != nil { message += fmt.Sprintf(" · first sweep %s · deterministic %d/%d · Luna %d · batches %d/%d+%d/%d", r.FirstSweep.Phase, r.FirstSweep.MechanicallyResolved, r.FirstSweep.ChangedTasks, r.FirstSweep.LunaCandidates, r.FirstSweep.FirstPassBatchesCompleted, r.FirstSweep.FirstPassBatchesTotal, r.FirstSweep.PreviousPassBatchesCompleted, r.FirstSweep.PreviousPassBatchesTotal) } @@ -171,18 +253,21 @@ func formatFailure(failure *state.Failure) string { } type InspectResult struct { - Version int `json:"version"` - TaskID string `json:"task_id"` - CapturedRevision string `json:"captured_revision"` - State state.TaskStatus `json:"state"` - Provenance state.Provenance `json:"provenance"` - ManagedAction string `json:"managed_action,omitempty"` - Retry *RetryResult `json:"retry,omitempty"` - ArchiveEligible bool `json:"archive_eligible"` - TokenDisplayPosition tokens.Position `json:"token_display_position"` - ManagedTokenPosition tokens.Position `json:"managed_token_position"` - ManagedTokenDisplay string `json:"managed_token_display"` - TokenUsageFound bool `json:"token_usage_found"` + Version int `json:"version"` + TaskID string `json:"task_id"` + CapturedRevision string `json:"captured_revision"` + State state.TaskStatus `json:"state"` + Provenance state.Provenance `json:"provenance"` + ManagedAction string `json:"managed_action,omitempty"` + Retry *RetryResult `json:"retry,omitempty"` + ArchiveEligible bool `json:"archive_eligible"` + TokenDisplayPosition tokens.Position `json:"token_display_position"` + ManagedTokenPosition tokens.Position `json:"managed_token_position"` + ManagedTokenDisplay string `json:"managed_token_display"` + TokenUsageFound bool `json:"token_usage_found"` + PendingTitlePlan bool `json:"pending_title_plan"` + NativeTitleOutcome state.NativeTitleOutcome `json:"native_title_outcome,omitempty"` + RenderedConvergence string `json:"rendered_convergence,omitempty"` } func (InspectResult) result() {} @@ -201,7 +286,7 @@ func (r InspectResult) Human() string { if r.Retry != nil { retry = fmt.Sprintf("%s/%s", r.Retry.Operation, r.Retry.ErrorCode) } - return fmt.Sprintf("%s %s · revision %s · provenance %s · next: %s · token configured %s · token applied %s/%s · token usage found %t · retry %s · archive eligible %t", r.State.Emoji(), r.TaskID, r.CapturedRevision, r.Provenance, action, r.TokenDisplayPosition, r.ManagedTokenPosition, display, r.TokenUsageFound, retry, r.ArchiveEligible) + return fmt.Sprintf("%s %s · revision %s · provenance %s · next: %s · token configured %s · token applied %s/%s · token usage found %t · retry %s · title pending %t/%s/%s · archive eligible %t", r.State.Emoji(), r.TaskID, r.CapturedRevision, r.Provenance, action, r.TokenDisplayPosition, r.ManagedTokenPosition, display, r.TokenUsageFound, retry, r.PendingTitlePlan, r.NativeTitleOutcome, r.RenderedConvergence, r.ArchiveEligible) } func (r InspectResult) normalized() InspectResult { @@ -650,6 +735,40 @@ func validateResult(value Result) error { if result.Allow || result.Disposition != "retired" { return errors.New("title dispatch compatibility envelope is invalid") } + case TitleBatchResult: + if result.Mode != "list" && result.Mode != "operation" && result.Mode != "stage" { + return errors.New("title batch mode is invalid") + } + seen := map[string]struct{}{} + for _, item := range result.Plans { + if err := checkID("operation_id", item.OperationID); err != nil { + return err + } + if err := checkID("task_id", item.TaskID); err != nil { + return err + } + if err := checkID("expected_revision", item.ExpectedRevision); err != nil { + return err + } + if item.DesiredTitle == "" { + return errors.New("desired title is required") + } + if _, ok := seen[item.OperationID]; ok { + return errors.New("duplicate title batch operation") + } + seen[item.OperationID] = struct{}{} + } + case TitleBatchReportResult: + seen := map[string]struct{}{} + for _, id := range append(append(append(slices.Clone(result.AcceptedIDs), result.FailedIDs...), result.DriftedIDs...), result.RejectedIDs...) { + if err := checkID("operation_id", id); err != nil { + return err + } + if _, ok := seen[id]; ok { + return errors.New("duplicate title report operation") + } + seen[id] = struct{}{} + } case HeartbeatResult: if result.Progress != nil { if err := result.Progress.Validate(); err != nil { diff --git a/internal/state/model.go b/internal/state/model.go index b3cb383..1b8728d 100644 --- a/internal/state/model.go +++ b/internal/state/model.go @@ -160,6 +160,9 @@ type PendingTitlePlan struct { NativeOutcome NativeTitleOutcome `json:"native_outcome"` NativeReportedAt *time.Time `json:"native_reported_at,omitempty"` NativeErrorCode string `json:"native_error_code,omitempty"` + ExpectedFooter string `json:"expected_footer,omitempty"` + CanonicalAttempts uint32 `json:"canonical_attempts,omitempty"` + CanonicalCheckedAt *time.Time `json:"canonical_checked_at,omitempty"` } func (p PendingTitlePlan) Validate() error { @@ -175,6 +178,12 @@ func (p PendingTitlePlan) Validate() error { if p.ManagedTokenDisplay != "" && p.ManagedTokenPosition != tokens.PositionStart && p.ManagedTokenPosition != tokens.PositionEnd { return errors.New("pending title token position is invalid") } + if strings.TrimSpace(p.ExpectedFooter) != p.ExpectedFooter { + return errors.New("pending title footer is invalid") + } + if p.CanonicalCheckedAt != nil && p.CanonicalCheckedAt.IsZero() { + return errors.New("pending title canonical check is invalid") + } switch p.NativeOutcome { case NativeTitlePending: if p.NativeReportedAt != nil || p.NativeErrorCode != "" { @@ -327,7 +336,8 @@ func (s State) Validate() error { return fmt.Errorf("pending title plan %s: %w", key, err) } task, ok := s.Tasks[key] - if !ok || task.CapturedRevision != plan.ExpectedRevision || task.CapturedTitle != plan.ExpectedTitle { + sourceAwaitingFooter := plan.ExpectedFooter != "" && plan.NativeOutcome == NativeTitleSucceeded && plan.CanonicalCheckedAt != nil + if !ok || !sourceAwaitingFooter && (task.CapturedRevision != plan.ExpectedRevision || task.CapturedTitle != plan.ExpectedTitle) { return fmt.Errorf("pending title plan %s does not match captured task", key) } } @@ -427,6 +437,7 @@ const ( StagePrepared OperationStage = "prepared" StageApplying OperationStage = "applying" StageApplied OperationStage = "applied" + StageNativePending OperationStage = "native_pending" StageVerified OperationStage = "verified" ) @@ -531,7 +542,7 @@ func (c CycleCheckpoint) Validate() error { } func (o CycleOperation) Valid() bool { - if o.Stage != StagePrepared && o.Stage != StageApplying && o.Stage != StageApplied && o.Stage != StageVerified { + if o.Stage != StagePrepared && o.Stage != StageApplying && o.Stage != StageApplied && o.Stage != StageNativePending && o.Stage != StageVerified { return false } switch o.Kind { diff --git a/internal/titlebatch/titlebatch.go b/internal/titlebatch/titlebatch.go new file mode 100644 index 0000000..cfbf026 --- /dev/null +++ b/internal/titlebatch/titlebatch.go @@ -0,0 +1,449 @@ +package titlebatch + +import ( + "context" + "encoding/json" + "errors" + "io" + "io/fs" + "sort" + "strings" + "time" + + "github.com/ericlitman/threadbear/internal/codex" + "github.com/ericlitman/threadbear/internal/config" + "github.com/ericlitman/threadbear/internal/output" + "github.com/ericlitman/threadbear/internal/state" + "github.com/ericlitman/threadbear/internal/status" + "github.com/ericlitman/threadbear/internal/title" + "github.com/ericlitman/threadbear/internal/tokens" +) + +const maxCanonicalAttempts = 3 + +type Store interface { + LoadConfig() (config.Config, error) + LoadState() (state.State, error) + SaveState(state.State) error + LoadCycle() (state.CycleCheckpoint, error) + AcquireLock() (*state.Lock, error) +} + +type Inventory interface { + Inventory(context.Context, string) (codex.Inventory, error) + Task(context.Context, string) (codex.Task, error) +} + +type Service struct { + Store Store + Inventory Inventory + Input io.Reader + Now func() time.Time + SourceIdentity func() string +} + +type stageEnvelope struct { + Footer string `json:"footer"` +} + +type nativeReport struct { + OperationID string `json:"operation_id"` + Outcome string `json:"outcome"` + ErrorCode string `json:"error_code,omitempty"` +} + +type reportEnvelope struct { + Reports []nativeReport `json:"reports"` +} + +func (s Service) List(ctx context.Context) (output.Result, error) { + return s.plan(ctx, "") +} + +func (s Service) Operation(ctx context.Context, operationID string) (output.Result, error) { + return s.plan(ctx, operationID) +} + +func (s Service) Stage(ctx context.Context) (output.Result, error) { + var envelope stageEnvelope + if err := decodeStrict(s.Input, &envelope); err != nil { + return commandError("invalid_stage", err) + } + lock, cfg, err := s.lockedConfig() + if err != nil { + return commandError(codeFor(err), err) + } + defer lock.Close() + if err := refuseCycle(s.Store); err != nil { + return commandError("cycle_in_progress", err) + } + committed, err := s.Store.LoadState() + if err != nil { + return commandError("state_read_failed", err) + } + task, err := s.Inventory.Task(ctx, cfg.ControlTaskID) + if err == nil && task.Archived { + err = errors.New("source task is archived") + } + if err != nil { + return commandError("source_missing", err) + } + record, owned := committed.Tasks[cfg.ControlTaskID] + if !owned { + record = state.TaskRecord{TaskID: task.TaskID, CapturedRevision: task.Revision, CapturedTitle: task.Title, Status: state.StatusUnknown, Provenance: state.ProvenanceUnknown, StateStartedAt: s.now(), LastSubstantiveActivity: s.now(), TokenDisplayPosition: cfg.TokenDisplay} + } + parsed := status.ParseFooter(status.FooterInput{Message: envelope.Footer, LatestTurnCompleted: true}) + if !parsed.Accepted { + return commandError("invalid_footer", errors.New("terminal footer is not accepted")) + } + record.CapturedRevision = task.Revision + record.CapturedTitle = task.Title + display := tokens.Display{} + if cfg.TokenDisplay != tokens.PositionOff && record.ManagedTokenDisplay != "" { + display = tokens.Display{Position: cfg.TokenDisplay, Value: record.ManagedTokenDisplay} + } + action := parsed.Footer.Action + if parsed.Footer.Owner == status.OwnerNone { + action = "" + } + rendered, err := title.Reconcile(record, parsed.Footer.Status, "", action, display) + if err != nil { + return commandError("title_render_failed", err) + } + record.Status, record.Provenance = parsed.Footer.Status, state.ProvenanceFooter + record.DurableSubject, record.ManagedAction = rendered.DurableSubject, rendered.ManagedAction + record.CapturedRevision, record.CapturedTitle = task.Revision, task.Title + committed.Tasks[task.TaskID] = record + plan := state.PendingTitlePlan{ + OperationID: state.TitleOperationID(task.TaskID, task.Revision, task.Title, rendered.Title), + TaskID: task.TaskID, ExpectedRevision: task.Revision, ExpectedTitle: task.Title, DesiredTitle: rendered.Title, + DurableSubject: rendered.DurableSubject, ManagedAction: rendered.ManagedAction, + ManagedTokenDisplay: rendered.ManagedTokenDisplay, ManagedTokenPosition: rendered.ManagedTokenPosition, + NativeOutcome: state.NativeTitlePending, ExpectedFooter: envelope.Footer, + } + if err := plan.Validate(); err != nil { + return commandError("invalid_plan", err) + } + committed.PendingTitlePlans[task.TaskID] = plan + committed.Generation++ + if err := s.Store.SaveState(committed); err != nil { + return commandError("state_write_failed", err) + } + return output.TitleBatchResult{Mode: "stage", Plans: []output.TitleBatchItem{}, Dispositions: []output.TitleBatchDisposition{{Outcome: "staged"}}}, nil +} + +func (s Service) Report() (output.Result, error) { + var envelope reportEnvelope + if err := decodeStrict(s.Input, &envelope); err != nil { + return commandError("invalid_report", err) + } + lock, _, err := s.lockedConfig() + if err != nil { + return commandError(codeFor(err), err) + } + defer lock.Close() + if err := refuseCycle(s.Store); err != nil { + return commandError("cycle_in_progress", err) + } + committed, err := s.Store.LoadState() + if err != nil { + return commandError("state_read_failed", err) + } + result := output.TitleBatchReportResult{AcceptedIDs: []string{}, FailedIDs: []string{}, DriftedIDs: []string{}, RejectedIDs: []string{}} + counts := map[string]int{} + for _, report := range envelope.Reports { + counts[report.OperationID]++ + } + now := s.now() + changed := false + for _, report := range envelope.Reports { + plan, taskID, ok := findPlan(committed, report.OperationID) + if !ok || report.OperationID == "" || counts[report.OperationID] != 1 || !validReport(report) { + if report.OperationID != "" { + result.RejectedIDs = appendUnique(result.RejectedIDs, report.OperationID) + } + continue + } + if plan.NativeOutcome == state.NativeTitleSucceeded { + if report.Outcome == "accepted" { + result.AcceptedIDs = append(result.AcceptedIDs, report.OperationID) + } else { + result.RejectedIDs = appendUnique(result.RejectedIDs, report.OperationID) + } + continue + } + candidate := plan + candidate.NativeReportedAt = &now + switch report.Outcome { + case "accepted": + candidate.NativeOutcome = state.NativeTitleSucceeded + candidate.NativeErrorCode = "" + candidate.CanonicalAttempts++ + result.AcceptedIDs = append(result.AcceptedIDs, report.OperationID) + case "failed": + candidate.NativeOutcome = state.NativeTitleFailed + candidate.NativeErrorCode = report.ErrorCode + result.FailedIDs = append(result.FailedIDs, report.OperationID) + case "drifted": + candidate.NativeOutcome = state.NativeTitleFailed + candidate.NativeErrorCode = "native_guard_drift" + result.DriftedIDs = append(result.DriftedIDs, report.OperationID) + } + if err := candidate.Validate(); err != nil { + result.RejectedIDs = appendUnique(result.RejectedIDs, report.OperationID) + continue + } + committed.PendingTitlePlans[taskID] = candidate + changed = true + } + if changed { + committed.Generation++ + if err := s.Store.SaveState(committed); err != nil { + return commandError("state_write_failed", err) + } + } + return result, nil +} + +func (s Service) plan(ctx context.Context, operationID string) (output.Result, error) { + lock, cfg, err := s.lockedConfig() + if err != nil { + return commandError(codeFor(err), err) + } + defer lock.Close() + if err := refuseCycle(s.Store); err != nil { + return commandError("cycle_in_progress", err) + } + committed, err := s.Store.LoadState() + if err != nil { + return commandError("state_read_failed", err) + } + observed, err := s.Inventory.Inventory(ctx, cfg.ControlTaskID) + if err != nil { + return commandError("inventory_failed", err) + } + if sourcePlan, pending := committed.PendingTitlePlans[cfg.ControlTaskID]; pending && sourcePlan.ExpectedFooter != "" { + sourceTask, readErr := s.Inventory.Task(ctx, cfg.ControlTaskID) + if readErr != nil { + return commandError("inventory_failed", readErr) + } + observed.Tasks = append(observed.Tasks, sourceTask) + } + mode := "list" + if operationID != "" { + mode = "operation" + } + result := output.TitleBatchResult{Mode: mode, Plans: []output.TitleBatchItem{}, Dispositions: []output.TitleBatchDisposition{}} + ids := make([]string, 0, len(committed.PendingTitlePlans)) + for taskID, plan := range committed.PendingTitlePlans { + if operationID == "" || plan.OperationID == operationID { + ids = append(ids, taskID) + } + } + sort.Strings(ids) + changed := false + now := s.now() + for _, taskID := range ids { + plan := committed.PendingTitlePlans[taskID] + task, exists := findTask(observed, taskID) + switch { + case exists && task.Archived: + delete(committed.PendingTitlePlans, taskID) + result.Dispositions = append(result.Dispositions, disposition(plan, "archived")) + changed = true + case !exists: + delete(committed.PendingTitlePlans, taskID) + result.Dispositions = append(result.Dispositions, disposition(plan, "missing")) + changed = true + case plan.ExpectedFooter != "" && task.Title == plan.DesiredTitle && plan.NativeOutcome == state.NativeTitleSucceeded: + if plan.CanonicalCheckedAt == nil { + plan.CanonicalCheckedAt = &now + committed.PendingTitlePlans[taskID] = plan + changed = true + } + result.Dispositions = append(result.Dispositions, disposition(plan, "canonical_verified_awaiting_footer")) + case plan.ExpectedFooter != "" && task.Revision != plan.ExpectedRevision: + result.Dispositions = append(result.Dispositions, disposition(plan, "awaiting_footer_verification")) + case task.Title == plan.DesiredTitle && plan.NativeOutcome == state.NativeTitleSucceeded: + applyCanonical(&committed, plan, task) + result.Dispositions = append(result.Dispositions, disposition(plan, "canonical_verified")) + changed = true + case task.Revision != plan.ExpectedRevision || task.Title != plan.ExpectedTitle: + delete(committed.PendingTitlePlans, taskID) + result.Dispositions = append(result.Dispositions, disposition(plan, "drifted")) + changed = true + case plan.NativeOutcome == state.NativeTitleSucceeded && plan.NativeReportedAt != nil && now.Before(plan.NativeReportedAt.Add(state.NativeTitleCanonicalTimeout)): + result.Dispositions = append(result.Dispositions, disposition(plan, "native_succeeded_pending_canonical")) + case plan.NativeOutcome == state.NativeTitleSucceeded && plan.CanonicalAttempts >= maxCanonicalAttempts: + result.Dispositions = append(result.Dispositions, disposition(plan, "canonical_verification_failed")) + case plan.NativeOutcome == state.NativeTitleSucceeded: + plan.NativeOutcome = state.NativeTitleFailed + plan.NativeErrorCode = "canonical_not_verified" + plan.CanonicalCheckedAt = &now + committed.PendingTitlePlans[taskID] = plan + result.Plans = append(result.Plans, item(plan)) + changed = true + default: + result.Plans = append(result.Plans, item(plan)) + } + } + if operationID != "" && len(ids) == 0 { + result.Dispositions = append(result.Dispositions, output.TitleBatchDisposition{OperationID: operationID, Outcome: "missing"}) + } + if changed { + committed.Generation++ + if err := s.Store.SaveState(committed); err != nil { + return commandError("state_write_failed", err) + } + } + return result, nil +} + +func (s Service) lockedConfig() (*state.Lock, config.Config, error) { + if s.Store == nil || s.Inventory == nil || s.Now == nil || s.SourceIdentity == nil { + return nil, config.Config{}, errors.New("dependency_unavailable") + } + lock, err := s.Store.AcquireLock() + if err != nil { + return nil, config.Config{}, err + } + cfg, err := s.Store.LoadConfig() + if err != nil { + lock.Close() + return nil, config.Config{}, err + } + sourceTaskID := s.SourceIdentity() + if !canonicalUUID(sourceTaskID) || sourceTaskID != cfg.ControlTaskID { + lock.Close() + return nil, config.Config{}, errors.New("source_identity_mismatch") + } + if !cfg.AgentsEnabled || !cfg.RenameEnabled { + lock.Close() + return nil, config.Config{}, errors.New("title_batch_disabled") + } + return lock, cfg, nil +} + +func refuseCycle(store Store) error { + _, err := store.LoadCycle() + if errors.Is(err, fs.ErrNotExist) { + return nil + } + if err != nil { + return err + } + return errors.New("cycle_in_progress") +} + +func decodeStrict(input io.Reader, target any) error { + if input == nil { + return errors.New("input unavailable") + } + decoder := json.NewDecoder(input) + decoder.DisallowUnknownFields() + if err := decoder.Decode(target); err != nil { + return err + } + if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) { + return errors.New("input must contain one JSON value") + } + return nil +} + +func validReport(report nativeReport) bool { + if report.OperationID == "" || strings.TrimSpace(report.OperationID) != report.OperationID { + return false + } + switch report.Outcome { + case "accepted", "drifted": + return report.ErrorCode == "" + case "failed": + return report.ErrorCode != "" && strings.TrimSpace(report.ErrorCode) == report.ErrorCode + default: + return false + } +} + +func item(plan state.PendingTitlePlan) output.TitleBatchItem { + return output.TitleBatchItem{OperationID: plan.OperationID, TaskID: plan.TaskID, ExpectedRevision: plan.ExpectedRevision, ExpectedTitle: plan.ExpectedTitle, DesiredTitle: plan.DesiredTitle} +} + +func disposition(plan state.PendingTitlePlan, outcome string) output.TitleBatchDisposition { + return output.TitleBatchDisposition{OperationID: plan.OperationID, TaskID: plan.TaskID, Outcome: outcome} +} + +func findTask(inventory codex.Inventory, taskID string) (codex.Task, bool) { + for _, task := range inventory.Tasks { + if task.TaskID == taskID { + return task, true + } + } + return codex.Task{}, false +} + +func findPlan(committed state.State, operationID string) (state.PendingTitlePlan, string, bool) { + for taskID, plan := range committed.PendingTitlePlans { + if plan.OperationID == operationID { + return plan, taskID, true + } + } + return state.PendingTitlePlan{}, "", false +} + +func applyCanonical(committed *state.State, plan state.PendingTitlePlan, task codex.Task) { + if record, ok := committed.Tasks[plan.TaskID]; ok { + record.CapturedRevision, record.CapturedTitle, record.LastAppliedTitle = task.Revision, task.Title, task.Title + record.DurableSubject, record.ManagedAction = plan.DurableSubject, plan.ManagedAction + record.ManagedTokenDisplay, record.ManagedTokenPosition = plan.ManagedTokenDisplay, plan.ManagedTokenPosition + committed.Tasks[plan.TaskID] = record + } + delete(committed.PendingTitlePlans, plan.TaskID) +} + +func appendUnique(values []string, value string) []string { + for _, existing := range values { + if existing == value { + return values + } + } + return append(values, value) +} + +func (s Service) now() time.Time { return s.Now().UTC() } + +func canonicalUUID(value string) bool { + if len(value) != 36 { + return false + } + for index, char := range value { + if index == 8 || index == 13 || index == 18 || index == 23 { + if char != '-' { + return false + } + continue + } + if char < '0' || char > '9' { + if char < 'a' || char > 'f' { + return false + } + } + } + return true +} + +func commandError(code string, err error) (output.Result, error) { + return output.ErrorResult{Operation: "title-batch", ErrorCode: code}, err +} + +func codeFor(err error) string { + switch err.Error() { + case "source_identity_mismatch": + return "source_identity_mismatch" + case "title_batch_disabled": + return "title_batch_disabled" + case "dependency_unavailable": + return "dependency_unavailable" + default: + return "title_batch_locked" + } +} diff --git a/internal/titlebatch/titlebatch_test.go b/internal/titlebatch/titlebatch_test.go new file mode 100644 index 0000000..343b09f --- /dev/null +++ b/internal/titlebatch/titlebatch_test.go @@ -0,0 +1,219 @@ +package titlebatch + +import ( + "bytes" + "context" + "errors" + "path/filepath" + "testing" + "time" + + "github.com/ericlitman/threadbear/internal/codex" + "github.com/ericlitman/threadbear/internal/config" + "github.com/ericlitman/threadbear/internal/output" + "github.com/ericlitman/threadbear/internal/state" +) + +const controlTaskID = "11111111-1111-4111-8111-111111111111" + +type fakeInventory struct{ tasks []codex.Task } + +func (f *fakeInventory) Inventory(_ context.Context, controlTaskID string) (codex.Inventory, error) { + tasks := make([]codex.Task, 0, len(f.tasks)) + for _, task := range f.tasks { + if task.TaskID != controlTaskID { + tasks = append(tasks, task) + } + } + return codex.Inventory{Tasks: tasks}, nil +} + +func (f *fakeInventory) Task(_ context.Context, taskID string) (codex.Task, error) { + for _, task := range f.tasks { + if task.TaskID == taskID { + return task, nil + } + } + return codex.Task{}, errors.New("missing task") +} + +func titleBatchFixture(t *testing.T, task codex.Task) (*state.Store, *fakeInventory, time.Time) { + t.Helper() + store := state.NewStore(filepath.Join(t.TempDir(), "state")) + cfg := config.Default(controlTaskID) + cfg.AgentsEnabled = true + if err := store.SaveConfig(cfg); err != nil { + t.Fatal(err) + } + committed := state.New() + committed.BootstrapComplete = true + committed.Tasks[task.TaskID] = state.TaskRecord{TaskID: task.TaskID, CapturedRevision: task.Revision, CapturedTitle: task.Title, LastAppliedTitle: task.Title, DurableSubject: "Control task", Status: state.StatusComplete, Provenance: state.ProvenanceFooter, StateStartedAt: time.Unix(1, 0), LastSubstantiveActivity: time.Unix(1, 0)} + if err := store.SaveState(committed); err != nil { + t.Fatal(err) + } + return store, &fakeInventory{tasks: []codex.Task{task}}, time.Unix(10, 0).UTC() +} + +func TestStageRequiresExactPersistedSourceIdentity(t *testing.T) { + task := codex.Task{TaskID: controlTaskID, Revision: "1", Title: "✅ Control task"} + store, inventory, now := titleBatchFixture(t, task) + service := Service{Store: store, Inventory: inventory, SourceIdentity: func() string { return "other" }, Input: bytes.NewBufferString(`{"footer":"🧵🐻 complete"}`), Now: func() time.Time { return now }} + result, err := service.Stage(context.Background()) + if err == nil || result.(output.ErrorResult).ErrorCode != "source_identity_mismatch" { + t.Fatalf("result=%+v err=%v", result, err) + } +} + +func TestStageListReportAndCanonicalSettlement(t *testing.T) { + task := codex.Task{TaskID: controlTaskID, Revision: "2", Title: "✅ Control task"} + store, inventory, now := titleBatchFixture(t, task) + service := Service{Store: store, Inventory: inventory, SourceIdentity: func() string { return controlTaskID }, Input: bytes.NewBufferString(`{"footer":"🧵🐻 needs input (you): choose the release region"}`), Now: func() time.Time { return now }} + staged, err := service.Stage(context.Background()) + if err != nil { + t.Fatal(err) + } + if got := staged.(output.TitleBatchResult); len(got.Plans) != 0 || len(got.Dispositions) != 1 || got.Dispositions[0].Outcome != "staged" { + t.Fatalf("staged=%+v", got) + } + stagedState, _ := store.LoadState() + storedPlan := stagedState.PendingTitlePlans[task.TaskID] + plan := output.TitleBatchItem{OperationID: storedPlan.OperationID, TaskID: storedPlan.TaskID, ExpectedRevision: storedPlan.ExpectedRevision, ExpectedTitle: storedPlan.ExpectedTitle, DesiredTitle: storedPlan.DesiredTitle} + if plan.DesiredTitle != "🙋 Control task → choose the release region" { + t.Fatalf("plan=%+v", plan) + } + listed, err := service.List(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(listed.(output.TitleBatchResult).Plans) != 1 { + t.Fatalf("listed=%+v", listed) + } + service.Input = bytes.NewBufferString(`{"reports":[{"operation_id":"` + plan.OperationID + `","outcome":"accepted"}]}`) + reported, err := service.Report() + if err != nil { + t.Fatal(err) + } + if len(reported.(output.TitleBatchReportResult).AcceptedIDs) != 1 { + t.Fatalf("reported=%+v", reported) + } + inventory.tasks[0].Title = plan.DesiredTitle + settled, err := service.List(context.Background()) + if err != nil { + t.Fatal(err) + } + if got := settled.(output.TitleBatchResult); len(got.Plans) != 0 || len(got.Dispositions) != 1 || got.Dispositions[0].Outcome != "canonical_verified_awaiting_footer" { + t.Fatalf("settled=%+v", got) + } + committed, _ := store.LoadState() + if committed.PendingTitlePlans[task.TaskID].CanonicalCheckedAt == nil || committed.Tasks[task.TaskID].LastAppliedTitle == plan.DesiredTitle { + t.Fatalf("state=%+v", committed) + } +} + +func TestOperationRevalidationRejectsDrift(t *testing.T) { + task := codex.Task{TaskID: "task-a", Revision: "1", Title: "Control task"} + store, inventory, now := titleBatchFixture(t, task) + committed, _ := store.LoadState() + plan := state.PendingTitlePlan{OperationID: state.TitleOperationID(task.TaskID, task.Revision, task.Title, "✅ Control task"), TaskID: task.TaskID, ExpectedRevision: task.Revision, ExpectedTitle: task.Title, DesiredTitle: "✅ Control task", DurableSubject: "Control task", NativeOutcome: state.NativeTitlePending} + committed.PendingTitlePlans[task.TaskID] = plan + if err := store.SaveState(committed); err != nil { + t.Fatal(err) + } + inventory.tasks[0].Revision = "2" + inventory.tasks[0].Title = "User edit" + service := Service{Store: store, Inventory: inventory, SourceIdentity: func() string { return controlTaskID }, Now: func() time.Time { return now }} + result, err := service.Operation(context.Background(), plan.OperationID) + if err != nil { + t.Fatal(err) + } + got := result.(output.TitleBatchResult) + if len(got.Plans) != 0 || len(got.Dispositions) != 1 || got.Dispositions[0].Outcome != "drifted" { + t.Fatalf("result=%+v", got) + } +} + +func TestNativeSuccessRetriesAfterBoundedCanonicalWait(t *testing.T) { + task := codex.Task{TaskID: "task-a", Revision: "1", Title: "Control task"} + store, inventory, now := titleBatchFixture(t, task) + committed, _ := store.LoadState() + plan := state.PendingTitlePlan{OperationID: state.TitleOperationID(task.TaskID, task.Revision, task.Title, "✅ Control task"), TaskID: task.TaskID, ExpectedRevision: task.Revision, ExpectedTitle: task.Title, DesiredTitle: "✅ Control task", DurableSubject: "Control task", NativeOutcome: state.NativeTitlePending} + committed.PendingTitlePlans[task.TaskID] = plan + if err := store.SaveState(committed); err != nil { + t.Fatal(err) + } + service := Service{Store: store, Inventory: inventory, SourceIdentity: func() string { return controlTaskID }, Input: bytes.NewBufferString(`{"reports":[{"operation_id":"` + plan.OperationID + `","outcome":"accepted"}]}`), Now: func() time.Time { return now }} + if _, err := service.Report(); err != nil { + t.Fatal(err) + } + pending, err := service.List(context.Background()) + if err != nil { + t.Fatal(err) + } + if pending.(output.TitleBatchResult).Dispositions[0].Outcome != "native_succeeded_pending_canonical" { + t.Fatalf("pending=%+v", pending) + } + now = now.Add(state.NativeTitleCanonicalTimeout) + retry, err := service.List(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(retry.(output.TitleBatchResult).Plans) != 1 { + t.Fatalf("retry=%+v", retry) + } +} + +func TestReportRejectsDuplicateOperationWithoutMutation(t *testing.T) { + task := codex.Task{TaskID: "task-a", Revision: "1", Title: "Control task"} + store, inventory, now := titleBatchFixture(t, task) + committed, _ := store.LoadState() + plan := state.PendingTitlePlan{OperationID: state.TitleOperationID(task.TaskID, task.Revision, task.Title, "✅ Control task"), TaskID: task.TaskID, ExpectedRevision: task.Revision, ExpectedTitle: task.Title, DesiredTitle: "✅ Control task", DurableSubject: "Control task", NativeOutcome: state.NativeTitlePending} + committed.PendingTitlePlans[task.TaskID] = plan + if err := store.SaveState(committed); err != nil { + t.Fatal(err) + } + body := `{"reports":[{"operation_id":"` + plan.OperationID + `","outcome":"accepted"},{"operation_id":"` + plan.OperationID + `","outcome":"accepted"}]}` + service := Service{Store: store, Inventory: inventory, SourceIdentity: func() string { return controlTaskID }, Input: bytes.NewBufferString(body), Now: func() time.Time { return now }} + result, err := service.Report() + if err != nil { + t.Fatal(err) + } + if len(result.(output.TitleBatchReportResult).RejectedIDs) != 1 { + t.Fatalf("result=%+v", result) + } + after, _ := store.LoadState() + if after.PendingTitlePlans[task.TaskID].NativeOutcome != state.NativeTitlePending { + t.Fatalf("state=%+v", after) + } +} + +func TestStageAdoptsRetainedSourceWithoutInferringActionOrTokenOwnership(t *testing.T) { + store := state.NewStore(filepath.Join(t.TempDir(), "state")) + cfg := config.Default(controlTaskID) + cfg.AgentsEnabled = true + if err := store.SaveConfig(cfg); err != nil { + t.Fatal(err) + } + if err := store.SaveState(state.New()); err != nil { + t.Fatal(err) + } + task := codex.Task{TaskID: controlTaskID, Revision: "1", Title: "✅ 26k User-owned control subject → user-owned suffix"} + inventory := &fakeInventory{tasks: []codex.Task{task}} + now := time.Unix(20, 0).UTC() + service := Service{Store: store, Inventory: inventory, SourceIdentity: func() string { return controlTaskID }, Input: bytes.NewBufferString(`{"footer":"🧵🐻 complete"}`), Now: func() time.Time { return now }} + result, err := service.Stage(context.Background()) + if err != nil { + t.Fatal(err) + } + if got := result.(output.TitleBatchResult); len(got.Plans) != 0 { + t.Fatalf("stage leaked manifest: %+v", got) + } + committed, _ := store.LoadState() + plan := committed.PendingTitlePlans[controlTaskID] + if plan.DesiredTitle != task.Title { + t.Fatalf("plan=%+v", plan) + } + record := committed.Tasks[controlTaskID] + if record.DurableSubject != "26k User-owned control subject → user-owned suffix" || record.ManagedAction != "" || record.ManagedTokenDisplay != "" { + t.Fatalf("record=%+v", record) + } +} diff --git a/internal/watch/cycle.go b/internal/watch/cycle.go index 1b9d62c..ccbde79 100644 --- a/internal/watch/cycle.go +++ b/internal/watch/cycle.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io/fs" + "slices" "sort" "strings" "time" @@ -185,6 +186,19 @@ func (r *Runner) Run(ctx context.Context, dryRun bool) (output.Result, error) { if err != nil { return output.HeartbeatResult{CycleID: "inventory", ErrorCode: "inventory_failed"}, err } + if plan, pending := committed.PendingTitlePlans[cfg.ControlTaskID]; pending && plan.ExpectedFooter != "" { + reader, ok := r.deps.Inventory.(interface { + Task(context.Context, string) (codex.Task, error) + }) + if !ok { + return output.HeartbeatResult{CycleID: "inventory", ErrorCode: "inventory_failed"}, errors.New("control task inventory is unavailable") + } + controlTask, readErr := reader.Task(ctx, cfg.ControlTaskID) + if readErr != nil { + return output.HeartbeatResult{CycleID: "inventory", ErrorCode: "inventory_failed"}, readErr + } + inventory.Tasks = append(inventory.Tasks, controlTask) + } titleStateChanged := false advancedTitleSettlements := make(map[string]struct{}) if !checkpointExists { @@ -203,15 +217,8 @@ func (r *Runner) Run(ctx context.Context, dryRun bool) (output.Result, error) { return output.HeartbeatResult{CycleID: "titles", ErrorCode: "state_write_failed"}, err } } - if !dryRun && !checkpointExists && cfg.RenameEnabled && hasPromotablePendingTitle(committed, inventory) { - checkpoint = state.NewCycle(r.deps.NewCycleID(), committed.Generation, now) - promotePendingTitles(committed, inventory, &checkpoint) - if err := r.deps.Store.SaveCycle(checkpoint); err != nil { - return output.HeartbeatResult{CycleID: checkpoint.CycleID, ErrorCode: "cycle_write_failed"}, err - } - checkpointExists = true - } comparison := codex.CompareInventory(inventory, committed) + comparison.RemovedIDs = slices.DeleteFunc(comparison.RemovedIDs, func(taskID string) bool { return taskID == cfg.ControlTaskID }) comparison.Changed = dueChanged(comparison.Changed, committed, now) archiveDue := archiveDueTasks(inventory, committed, cfg, now) comparison.Changed = mergeChanged(comparison.Changed, archiveDue) @@ -979,6 +986,9 @@ func (r *Runner) recoverOperations(checkpoint *state.CycleCheckpoint, inventory task, exists := findTask(inventory, operation.TaskID) switch operation.Kind { case state.OperationTitle: + if operation.Stage == state.StageNativePending { + continue + } if exists && task.Title == operation.DesiredTitle && (!operation.ForceWrite || operation.Stage == state.StageApplied) { operation.Stage = state.StageVerified operation.VerifiedRevision = task.Revision @@ -1004,6 +1014,11 @@ func (r *Runner) prepareRecords(cfg config.Config, committed state.State, checkp for taskID, captured := range checkpoint.Inventory { classification, ok := checkpoint.Results[taskID] previous, hadPrevious := committed.Tasks[taskID] + if plan, pending := committed.PendingTitlePlans[taskID]; pending && hadPrevious && plan.ExpectedFooter != "" && plan.NativeOutcome == state.NativeTitleSucceeded && captured.Title == plan.DesiredTitle { + previous.CapturedRevision, previous.CapturedTitle, previous.LastAppliedTitle = captured.Revision, captured.Title, captured.Title + previous.DurableSubject, previous.ManagedAction = plan.DurableSubject, plan.ManagedAction + previous.ManagedTokenDisplay, previous.ManagedTokenPosition = plan.ManagedTokenDisplay, plan.ManagedTokenPosition + } if !ok { if hadPrevious { previous.CapturedRevision = captured.Revision @@ -1281,7 +1296,7 @@ func (r *Runner) applyOperations(ctx context.Context, cfg config.Config, client if op.Kind == state.OperationArchive && hasPendingTitleOperation(*checkpoint, op.TaskID) { continue } - if client == nil { + if client == nil && op.Kind != state.OperationTitle { r.setDiagnostic(checkpoint, op.TaskID, string(op.Kind), "app_server_unavailable") continue } @@ -1295,50 +1310,10 @@ func (r *Runner) applyOperations(ctx context.Context, cfg config.Config, client } switch op.Kind { case state.OperationTitle: - op.Stage = state.StageApplying - checkpoint.Operations[key] = op - if err := r.deps.Store.SaveCycle(*checkpoint); err != nil { - return err - } - if err := client.SetTitle(ctx, op.TaskID, op.DesiredTitle); err != nil { - op.Stage = state.StagePrepared - checkpoint.Operations[key] = op - r.setDiagnostic(checkpoint, op.TaskID, "title", "title_write_failed") - continue - } - op.Stage = state.StageApplied - checkpoint.Operations[key] = op - if err := r.deps.Store.SaveCycle(*checkpoint); err != nil { - return err - } - verified, verifyErr := r.deps.Inventory.Inventory(ctx, cfg.ControlTaskID) - if verifyErr != nil { - return verifyErr - } - row, ok := findTask(verified, op.TaskID) - if !ok || row.Title != op.DesiredTitle { - return errors.New("title mutation was not visible after application") - } - op.Stage = state.StageVerified - op.VerifiedRevision = row.Revision - op.VerifiedTitle = row.Title + op.Stage = state.StageNativePending checkpoint.Operations[key] = op record := records[op.TaskID] - record.CapturedRevision = row.Revision - record.CapturedTitle = row.Title - record.LastAppliedTitle = row.Title - record.DurableSubject = op.DurableSubject - record.ManagedAction = op.ManagedAction - record.ManagedTokenDisplay = op.ManagedTokenDisplay - record.ManagedTokenPosition = op.ManagedTokenPosition - records[op.TaskID] = record result.Changed = append(result.Changed, output.TaskChange{TaskID: op.TaskID, State: record.Status}) - archiveKey := "archive:" + op.TaskID - if archive, exists := checkpoint.Operations[archiveKey]; exists && archive.Stage == state.StagePrepared { - archive.ExpectedRevision = row.Revision - archive.ExpectedTitle = row.Title - checkpoint.Operations[archiveKey] = archive - } if err := r.deps.Store.SaveCycle(*checkpoint); err != nil { return err } @@ -1418,8 +1393,10 @@ func settleOrDrainPendingTitles(committed *state.State, inventory codex.Inventor changed = true continue } - sameTitleRefresh := plan.ExpectedTitle == plan.DesiredTitle - if task.Title == plan.DesiredTitle && (!sameTitleRefresh || plan.NativeOutcome == state.NativeTitleSucceeded) { + if plan.ExpectedFooter != "" { + continue + } + if task.Title == plan.DesiredTitle && plan.NativeOutcome == state.NativeTitleSucceeded { revisionAdvanced := task.Revision != plan.ExpectedRevision if record, ok := committed.Tasks[taskID]; ok { if !revisionAdvanced { @@ -1445,30 +1422,6 @@ func settleOrDrainPendingTitles(committed *state.State, inventory codex.Inventor return changed, advanced } -func hasPromotablePendingTitle(committed state.State, inventory codex.Inventory) bool { - for taskID, plan := range committed.PendingTitlePlans { - task, exists := findTask(inventory, taskID) - _, recorded := committed.Tasks[taskID] - if exists && recorded && task.Revision == plan.ExpectedRevision && task.Title == plan.ExpectedTitle { - return true - } - } - return false -} - -func promotePendingTitles(committed state.State, inventory codex.Inventory, checkpoint *state.CycleCheckpoint) { - for taskID, plan := range committed.PendingTitlePlans { - task, exists := findTask(inventory, taskID) - record, recorded := committed.Tasks[taskID] - if !exists || !recorded || task.Revision != plan.ExpectedRevision || task.Title != plan.ExpectedTitle { - continue - } - checkpoint.Inventory[taskID] = state.CapturedTask{TaskID: taskID, Revision: task.Revision, Title: task.Title, RolloutPath: task.RolloutPath, Archived: task.Archived, LastSubstantiveActivity: record.LastSubstantiveActivity, EvidenceFingerprint: record.EvidenceFingerprint} - checkpoint.Results[taskID] = state.ClassificationResult{TaskID: taskID, Revision: task.Revision, Status: record.Status, Provenance: record.Provenance, DurableSubject: plan.DurableSubject, ManagedAction: plan.ManagedAction} - checkpoint.Operations["title:"+taskID] = state.CycleOperation{Kind: state.OperationTitle, Stage: state.StagePrepared, TaskID: taskID, ExpectedRevision: plan.ExpectedRevision, ExpectedTitle: plan.ExpectedTitle, DesiredTitle: plan.DesiredTitle, DurableSubject: plan.DurableSubject, ManagedAction: plan.ManagedAction, ManagedTokenDisplay: plan.ManagedTokenDisplay, ManagedTokenPosition: plan.ManagedTokenPosition, ForceWrite: true} - } -} - func incompleteEvidence(diagnostic state.CycleDiagnostic) bool { if diagnostic.Operation != "evidence" { return false @@ -1506,6 +1459,20 @@ func (r *Runner) commitState(cfg config.Config, committed state.State, checkpoin } continue } + if operation.Stage == state.StageNativePending { + plan := state.PendingTitlePlan{ + OperationID: state.TitleOperationID(operation.TaskID, operation.ExpectedRevision, operation.ExpectedTitle, operation.DesiredTitle), + TaskID: operation.TaskID, ExpectedRevision: operation.ExpectedRevision, ExpectedTitle: operation.ExpectedTitle, DesiredTitle: operation.DesiredTitle, + DurableSubject: operation.DurableSubject, ManagedAction: operation.ManagedAction, + ManagedTokenDisplay: operation.ManagedTokenDisplay, ManagedTokenPosition: operation.ManagedTokenPosition, + NativeOutcome: state.NativeTitlePending, + } + if err := plan.Validate(); err != nil { + return state.State{}, fmt.Errorf("stage native title plan for %s: %w", operation.TaskID, err) + } + next.PendingTitlePlans[operation.TaskID] = plan + continue + } if operation.Stage == state.StageVerified { if record, ok := records[operation.TaskID]; ok { if legacyTitleOwnership(operation) { @@ -1534,6 +1501,31 @@ func (r *Runner) commitState(cfg config.Config, committed state.State, checkpoin for taskID, record := range records { next.Tasks[taskID] = record } + if controlRecord, ok := committed.Tasks[cfg.ControlTaskID]; ok { + if _, captured := next.Tasks[cfg.ControlTaskID]; !captured { + next.Tasks[cfg.ControlTaskID] = controlRecord + } + } + for taskID, plan := range next.PendingTitlePlans { + if plan.ExpectedFooter == "" { + continue + } + classification, ok := checkpoint.Results[taskID] + record, recorded := next.Tasks[taskID] + parsed := status.ParseFooter(status.FooterInput{Message: plan.ExpectedFooter, LatestTurnCompleted: true}) + action := "" + if parsed.Accepted && parsed.Footer.Owner != status.OwnerNone { + action = parsed.Footer.Action + } + if !ok || !recorded || !parsed.Accepted || classification.Status != parsed.Footer.Status || classification.ManagedAction != action || record.CapturedTitle != plan.DesiredTitle || plan.NativeOutcome != state.NativeTitleSucceeded || plan.CanonicalCheckedAt == nil { + continue + } + record.LastAppliedTitle = plan.DesiredTitle + record.DurableSubject, record.ManagedAction = plan.DurableSubject, plan.ManagedAction + record.ManagedTokenDisplay, record.ManagedTokenPosition = plan.ManagedTokenDisplay, plan.ManagedTokenPosition + next.Tasks[taskID] = record + delete(next.PendingTitlePlans, taskID) + } for taskID := range next.Archives { if _, restored := checkpoint.Inventory[taskID]; restored { delete(next.Archives, taskID) diff --git a/internal/watch/cycle_test.go b/internal/watch/cycle_test.go index f14a85e..959e69e 100644 --- a/internal/watch/cycle_test.go +++ b/internal/watch/cycle_test.go @@ -35,7 +35,7 @@ type fakeIndex struct { events *[]string } -func (f *fakeIndex) Inventory(context.Context, string) (codex.Inventory, error) { +func (f *fakeIndex) Inventory(_ context.Context, controlTaskID string) (codex.Inventory, error) { f.calls++ if f.events != nil { *f.events = append(*f.events, "inventory") @@ -43,7 +43,21 @@ func (f *fakeIndex) Inventory(context.Context, string) (codex.Inventory, error) if f.hook != nil { f.hook(f.calls, f) } - return codex.Inventory{Tasks: append([]codex.Task{}, f.tasks...)}, nil + tasks := make([]codex.Task, 0, len(f.tasks)) + for _, task := range f.tasks { + if task.TaskID != controlTaskID { + tasks = append(tasks, task) + } + } + return codex.Inventory{Tasks: tasks}, nil +} + +func (f *fakeIndex) Task(_ context.Context, taskID string) (codex.Task, error) { + task, ok := f.task(taskID) + if !ok { + return codex.Task{}, errors.New("missing task") + } + return task, nil } func (f *fakeIndex) task(taskID string) (codex.Task, bool) { @@ -457,31 +471,21 @@ func TestHeartbeatRendersOutputTokensAndLeavesUnchangedTitlesAlone(t *testing.T) previous.TokenDisplayPosition = tokens.PositionStart committed.Tasks[task.TaskID] = previous runner, deps := testRunner(t, now, []codex.Task{task}, committed) - deps.client.latest[task.TaskID] = appserver.RecentEvidence{ - ThreadStatus: appserver.ThreadStatus{Type: "idle"}, - Latest: &appserver.EvidenceTurn{ID: "turn-a", Status: "failed", Error: &appserver.TurnError{Message: "synthetic failure"}}, - } - deps.tokens.snapshots[task.RolloutPath] = tokens.Snapshot{ - RolloutPath: task.RolloutPath, Offset: 120, Size: 120, OutputTokens: 1_600_123, TotalTokens: 433_000_000, Found: true, - } - + deps.client.latest[task.TaskID] = appserver.RecentEvidence{ThreadStatus: appserver.ThreadStatus{Type: "idle"}, Latest: &appserver.EvidenceTurn{ID: "turn-a", Status: "failed", Error: &appserver.TurnError{Message: "synthetic failure"}}} + deps.tokens.snapshots[task.RolloutPath] = tokens.Snapshot{RolloutPath: task.RolloutPath, Offset: 120, Size: 120, OutputTokens: 1_600_123, TotalTokens: 433_000_000, Found: true} if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } - current, _ := deps.index.task(task.TaskID) stored, _ := deps.store.store.LoadState() - if current.Title != "🚨 1.6m Release service" || stored.Tasks[task.TaskID].CapturedTitle != current.Title || len(stored.PendingTitlePlans) != 0 || len(deps.client.titles) != 1 || len(deps.tokens.calls) != 1 { - t.Fatalf("title=%q state=%+v writes=%v token_reads=%v", current.Title, stored.Tasks[task.TaskID], deps.client.titles, deps.tokens.calls) - } - - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) + plan := stored.PendingTitlePlans[task.TaskID] + if plan.DesiredTitle != "🚨 1.6m Release service" || plan.NativeOutcome != state.NativeTitlePending || len(deps.client.titles) != 0 || len(deps.tokens.calls) != 1 { + t.Fatalf("plan=%+v writes=%v reads=%v", plan, deps.client.titles, deps.tokens.calls) } if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } - if len(deps.client.titles) != 1 || len(deps.tokens.calls) != 1 { - t.Fatalf("unchanged heartbeats wrote titles or reread tokens: writes=%v reads=%v", deps.client.titles, deps.tokens.calls) + if len(deps.client.titles) != 0 || len(deps.tokens.calls) != 1 { + t.Fatalf("writes=%v reads=%v", deps.client.titles, deps.tokens.calls) } } @@ -537,17 +541,14 @@ func TestHeartbeatRepositionsTokenDisplayWithoutReclassification(t *testing.T) { cfg := config.Default("control") cfg.TokenDisplay = tokens.PositionEnd deps.store.configOverride = &cfg - deps.tokens.snapshots[task.RolloutPath] = tokens.Snapshot{ - RolloutPath: task.RolloutPath, Offset: 120, Size: 120, OutputTokens: 1_600_123, TotalTokens: 433_000_000, Found: true, - } - + deps.tokens.snapshots[task.RolloutPath] = tokens.Snapshot{RolloutPath: task.RolloutPath, Offset: 120, Size: 120, OutputTokens: 1_600_123, Found: true} if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } - current, _ := deps.index.task(task.TaskID) stored, _ := deps.store.store.LoadState() - if current.Title != "➡️ Release service → review rollout · 1.6m" || stored.Tasks[task.TaskID].CapturedTitle != current.Title || len(stored.PendingTitlePlans) != 0 || deps.classifier.calls != 0 || len(deps.client.latestReads) != 0 || len(deps.client.titles) != 1 { - t.Fatalf("title=%q state=%+v classifier=%d latest_reads=%v writes=%v", current.Title, stored.Tasks[task.TaskID], deps.classifier.calls, deps.client.latestReads, deps.client.titles) + plan := stored.PendingTitlePlans[task.TaskID] + if plan.DesiredTitle != "➡️ Release service → review rollout · 1.6m" || deps.classifier.calls != 0 || len(deps.client.latestReads) != 0 || len(deps.client.titles) != 0 { + t.Fatalf("plan=%+v classifier=%d reads=%v writes=%v", plan, deps.classifier.calls, deps.client.latestReads, deps.client.titles) } } @@ -571,29 +572,14 @@ func TestHeartbeatCleansRepeatedOwnedPrefixWhenMovingDisplayToEnd(t *testing.T) cfg := config.Default("control") cfg.TokenDisplay = tokens.PositionEnd deps.store.configOverride = &cfg - deps.tokens.snapshots[task.RolloutPath] = tokens.Snapshot{ - RolloutPath: task.RolloutPath, Offset: 120, Size: 120, OutputTokens: 26_123, TotalTokens: 433_000_000, Found: true, - } - + deps.tokens.snapshots[task.RolloutPath] = tokens.Snapshot{RolloutPath: task.RolloutPath, Offset: 120, Size: 120, OutputTokens: 26_123, Found: true} if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } - current, _ := deps.index.task(task.TaskID) - stored, err := deps.store.store.LoadState() - if err != nil { - t.Fatal(err) - } - owned := stored.Tasks[task.TaskID] - if current.Title != "✅ 26k 26k Execute BEAR-59 · 26k" || owned.DurableSubject != "26k 26k Execute BEAR-59" || owned.ManagedTokenDisplay != "26k" || owned.ManagedTokenPosition != tokens.PositionEnd || len(stored.PendingTitlePlans) != 0 || len(deps.client.titles) != 1 { - t.Fatalf("title=%q state=%+v writes=%v", current.Title, owned, deps.client.titles) - } - - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) - } - current, _ = deps.index.task(task.TaskID) - if current.Title != "✅ 26k 26k Execute BEAR-59 · 26k" || len(deps.client.titles) != 1 { - t.Fatalf("second title=%q writes=%v", current.Title, deps.client.titles) + stored, _ := deps.store.store.LoadState() + plan := stored.PendingTitlePlans[task.TaskID] + if plan.DesiredTitle != "✅ 26k 26k Execute BEAR-59 · 26k" || plan.DurableSubject != "26k 26k Execute BEAR-59" || plan.ManagedTokenDisplay != "26k" || plan.ManagedTokenPosition != tokens.PositionEnd || len(deps.client.titles) != 0 { + t.Fatalf("plan=%+v writes=%v", plan, deps.client.titles) } } @@ -621,7 +607,7 @@ func TestHeartbeatUnreadableTokenTailRemovesFigureWithoutRetryNoise(t *testing.T } current, _ := deps.index.task(task.TaskID) stored, _ := deps.store.store.LoadState() - if current.Title != "🚨 Release service" || stored.Tasks[task.TaskID].CapturedTitle != current.Title || len(stored.PendingTitlePlans) != 0 || len(value.(output.HeartbeatResult).Retries) != 0 || len(deps.client.titles) != 1 { + if current.Title != task.Title || stored.PendingTitlePlans[task.TaskID].DesiredTitle != "🚨 Release service" || len(value.(output.HeartbeatResult).Retries) != 0 || len(deps.client.titles) != 0 { t.Fatalf("title=%q state=%+v result=%+v writes=%v", current.Title, stored.Tasks[task.TaskID], value, deps.client.titles) } } @@ -658,39 +644,33 @@ func TestHeartbeatMissingRolloutPathRemovesFigureWithoutRetryNoise(t *testing.T) t.Fatal(err) } record := stored.Tasks[task.TaskID] - if current.Title != "🚨 Release service" || record.CapturedTitle != current.Title || len(stored.PendingTitlePlans) != 0 || len(value.(output.HeartbeatResult).Retries) != 0 || len(deps.tokens.calls) != 0 || len(deps.client.titles) != 1 { + if current.Title != task.Title || stored.PendingTitlePlans[task.TaskID].DesiredTitle != "🚨 Release service" || len(value.(output.HeartbeatResult).Retries) != 0 || len(deps.tokens.calls) != 0 || len(deps.client.titles) != 0 { t.Fatalf("title=%q state=%+v result=%+v token_reads=%v writes=%v", current.Title, record, value, deps.tokens.calls, deps.client.titles) } - if record.ManagedTokenDisplay != "" || record.ManagedTokenPosition != "" || record.TokenRolloutPath != "" || record.TokenReadOffset != 0 || record.TokenRolloutSize != 0 || record.OutputTokens != 0 || record.TotalTokens != 0 || record.TokenUsageFound { + if record.ManagedTokenDisplay != "1.6m" || record.ManagedTokenPosition != tokens.PositionStart || record.TokenRolloutPath != "" || record.TokenReadOffset != 0 || record.TokenRolloutSize != 0 || record.OutputTokens != 0 || record.TotalTokens != 0 || record.TokenUsageFound { t.Fatalf("stale token snapshot retained: %+v", record) } } func TestHeartbeatDeterministicRuntimeAndPartialSiblingFailure(t *testing.T) { now := time.Date(2026, 7, 24, 12, 0, 0, 0, time.UTC) - tasks := []codex.Task{ - {TaskID: "task-a", Revision: "2", Title: "✅ Active work", Source: "vscode"}, - {TaskID: "task-b", Revision: "2", Title: "⏳ Finished work", Source: "vscode"}, - } + tasks := []codex.Task{{TaskID: "task-a", Revision: "2", Title: "✅ Active work", Source: "vscode"}, {TaskID: "task-b", Revision: "2", Title: "⏳ Finished work", Source: "vscode"}} committed := state.New() committed.LastUpdateCheck = timePointer(now) committed.Tasks["task-a"] = record(codex.Task{TaskID: "task-a", Revision: "1", Title: "✅ Active work"}, state.StatusComplete, now) committed.Tasks["task-b"] = record(codex.Task{TaskID: "task-b", Revision: "1", Title: "✅ Finished work"}, state.StatusComplete, now) runner, deps := testRunner(t, now, tasks, committed) activeAt := now.Unix() - deps.client.latest["task-a"] = appserver.RecentEvidence{ThreadStatus: appserver.ThreadStatus{Type: "active"}, RecencyAt: &activeAt, Latest: &appserver.EvidenceTurn{ID: "turn-a", Status: "inProgress", UserMessage: "continue"}} + deps.client.latest["task-a"] = appserver.RecentEvidence{ThreadStatus: appserver.ThreadStatus{Type: "active"}, RecencyAt: &activeAt, Latest: &appserver.EvidenceTurn{ID: "a", Status: "inProgress"}} deps.client.latest["task-b"] = completedEvidence(now, "done", "🧵🐻 complete") value, err := runner.Run(context.Background(), false) if err != nil { t.Fatal(err) } result := value.(output.HeartbeatResult) - if deps.classifier.calls != 0 || len(result.Changed) != 2 || len(result.Retries) != 0 { - t.Fatalf("result=%+v classifier=%d", result, deps.classifier.calls) - } - stored, loadErr := deps.store.store.LoadState() - if loadErr != nil || len(stored.PendingTitlePlans) != 0 || stored.Tasks["task-a"].CapturedTitle != "⏳ Active work" || stored.Tasks["task-b"].CapturedTitle != "✅ Finished work" || stored.Tasks["task-a"].Status != state.StatusRunning { - t.Fatalf("state=%+v err=%v", stored, loadErr) + stored, _ := deps.store.store.LoadState() + if deps.classifier.calls != 0 || len(result.Changed) != 2 || stored.Tasks["task-a"].Status != state.StatusRunning || stored.PendingTitlePlans["task-a"].DesiredTitle != "⏳ Active work" || stored.PendingTitlePlans["task-b"].DesiredTitle != "✅ Finished work" { + t.Fatalf("result=%+v state=%+v", result, stored) } } @@ -801,20 +781,19 @@ func TestCycleRemovesOwnedActionWhenDeterministicStateHasNone(t *testing.T) { task := codex.Task{TaskID: "active", Revision: "2", Title: "➡️ Ship release → deploy production", Source: "vscode"} committed := state.New() committed.LastUpdateCheck = timePointer(now) - record := record(codex.Task{TaskID: task.TaskID, Revision: "1", Title: task.Title}, state.StatusNextSteps, now) - record.DurableSubject = "Ship release" - record.ManagedAction = "deploy production" - committed.Tasks[task.TaskID] = record + rec := record(codex.Task{TaskID: task.TaskID, Revision: "1", Title: task.Title}, state.StatusNextSteps, now) + rec.DurableSubject = "Ship release" + rec.ManagedAction = "deploy production" + committed.Tasks[task.TaskID] = rec runner, deps := testRunner(t, now, []codex.Task{task}, committed) activeAt := now.Unix() deps.client.latest[task.TaskID] = appserver.RecentEvidence{ThreadStatus: appserver.ThreadStatus{Type: "active"}, RecencyAt: &activeAt, Latest: &appserver.EvidenceTurn{ID: "turn", Status: "inProgress"}} if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } - current, _ := deps.index.task(task.TaskID) stored, _ := deps.store.store.LoadState() - if current.Title != "⏳ Ship release" || stored.Tasks[task.TaskID].CapturedTitle != current.Title || len(stored.PendingTitlePlans) != 0 { - t.Fatalf("title=%q state=%+v", current.Title, stored.Tasks[task.TaskID]) + if stored.PendingTitlePlans[task.TaskID].DesiredTitle != "⏳ Ship release" { + t.Fatalf("state=%+v", stored) } } @@ -932,63 +911,25 @@ func TestCrashAfterClassifierReusesCheckpoint(t *testing.T) { func TestCrashAfterTitleAndArchiveResumeVerifiedOperations(t *testing.T) { now := time.Date(2026, 7, 24, 12, 0, 0, 0, time.UTC) - t.Run("title", func(t *testing.T) { - tasks := []codex.Task{{TaskID: "semantic", Revision: "2", Title: "Semantic", Source: "vscode"}} - committed := state.New() - committed.LastUpdateCheck = timePointer(now) - committed.Tasks["semantic"] = record(codex.Task{TaskID: "semantic", Revision: "1", Title: "Semantic"}, state.StatusUnknown, now.Add(-time.Hour)) - runner, deps := testRunner(t, now, tasks, committed) - deps.client.latest["semantic"] = completedEvidence(now, "done", "no footer") - deps.classifier.results["semantic"] = status.Classification{TaskID: "semantic", Revision: "2", Status: state.StatusComplete, Provenance: state.ProvenanceLuna, DurableSubject: "Semantic"} - deps.store.failSaveState = true - if _, err := runner.Run(context.Background(), false); err == nil { - t.Fatal("first run did not crash") - } - if len(deps.client.titles) != 1 || deps.classifier.calls != 1 { - t.Fatalf("titles=%v classifier=%d", deps.client.titles, deps.classifier.calls) - } - deps.store.failSaveState = false - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) - } - if len(deps.client.titles) != 1 || deps.classifier.calls != 1 { - t.Fatalf("operation repeated: titles=%v classifier=%d", deps.client.titles, deps.classifier.calls) - } - stored, _ := deps.store.store.LoadState() - current, _ := deps.index.task("semantic") - if stored.Tasks["semantic"].CapturedRevision != current.Revision || stored.Tasks["semantic"].CapturedTitle != current.Title { - t.Fatalf("stored=%+v current=%+v", stored.Tasks["semantic"], current) - } - }) - t.Run("archive", func(t *testing.T) { - task := codex.Task{TaskID: "complete", Revision: "1", Title: "✅ Complete", Source: "vscode"} - committed := state.New() - committed.LastUpdateCheck = timePointer(now) - record := record(task, state.StatusComplete, now.Add(-15*24*time.Hour)) - record.StateStartedAt = now.Add(-15 * 24 * time.Hour) - committed.Tasks[task.TaskID] = record - runner, deps := testRunner(t, now, []codex.Task{task}, committed) - oldActivity := now.Add(-15 * 24 * time.Hour).Unix() - deps.client.latest[task.TaskID] = appserver.RecentEvidence{ThreadStatus: appserver.ThreadStatus{Type: "idle"}, RecencyAt: &oldActivity, Latest: &appserver.EvidenceTurn{ID: "turn", Status: "completed"}} - deps.store.failSaveState = true - if _, err := runner.Run(context.Background(), false); err == nil { - t.Fatal("first run did not crash") - } - if len(deps.client.archives) != 1 { - t.Fatalf("archives=%v", deps.client.archives) - } - deps.store.failSaveState = false - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) - } - if len(deps.client.archives) != 1 { - t.Fatalf("archive repeated: %v", deps.client.archives) - } - stored, _ := deps.store.store.LoadState() - if _, ok := stored.Archives[task.TaskID]; !ok { - t.Fatal("archive ownership missing after recovery") - } - }) + task := codex.Task{TaskID: "semantic", Revision: "2", Title: "Semantic", Source: "vscode"} + committed := state.New() + committed.LastUpdateCheck = timePointer(now) + committed.Tasks[task.TaskID] = record(codex.Task{TaskID: task.TaskID, Revision: "1", Title: "Semantic"}, state.StatusUnknown, now) + runner, deps := testRunner(t, now, []codex.Task{task}, committed) + deps.client.latest[task.TaskID] = completedEvidence(now, "done", "no footer") + deps.classifier.results[task.TaskID] = status.Classification{TaskID: task.TaskID, Revision: "2", Status: state.StatusComplete, Provenance: state.ProvenanceLuna, DurableSubject: "Semantic"} + deps.store.failSaveState = true + if _, err := runner.Run(context.Background(), false); err == nil { + t.Fatal("first run did not crash") + } + deps.store.failSaveState = false + if _, err := runner.Run(context.Background(), false); err != nil { + t.Fatal(err) + } + stored, _ := deps.store.store.LoadState() + if stored.PendingTitlePlans[task.TaskID].DesiredTitle != "✅ Semantic" || deps.classifier.calls != 1 || len(deps.client.titles) != 0 { + t.Fatalf("state=%+v classifier=%d writes=%v", stored, deps.classifier.calls, deps.client.titles) + } } func TestCrashBeforeArchiveDoesNotClaimExternalArchive(t *testing.T) { @@ -1022,162 +963,54 @@ func TestCrashBeforeArchiveDoesNotClaimExternalArchive(t *testing.T) { func TestCrashApplyingJournalRecoversWithoutRepeatingMutation(t *testing.T) { now := time.Date(2026, 7, 24, 12, 0, 0, 0, time.UTC) - t.Run("title already applied", func(t *testing.T) { - current := codex.Task{TaskID: "title", Revision: "2", Title: "✅ Title", Source: "vscode"} - committed := state.New() - committed.LastUpdateCheck = timePointer(now) - committed.Tasks[current.TaskID] = record(codex.Task{TaskID: current.TaskID, Revision: "1", Title: "Title"}, state.StatusComplete, now) - runner, deps := testRunner(t, now, []codex.Task{current}, committed) - cycle := state.NewCycle("cycle-1", committed.Generation, now) - cycle.Inventory[current.TaskID] = state.CapturedTask{TaskID: current.TaskID, Revision: "1", Title: "Title", LastSubstantiveActivity: now} - cycle.Results[current.TaskID] = state.ClassificationResult{TaskID: current.TaskID, Revision: "1", Status: state.StatusComplete, Provenance: state.ProvenanceFooter, DurableSubject: "Title"} - cycle.Operations["title:title"] = state.CycleOperation{Kind: state.OperationTitle, Stage: state.StageApplying, TaskID: current.TaskID, ExpectedRevision: "1", ExpectedTitle: "Title", DesiredTitle: current.Title} - if err := deps.store.store.SaveCycle(cycle); err != nil { - t.Fatal(err) - } - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) - } - if len(deps.client.titles) != 0 { - t.Fatalf("title repeated: %v", deps.client.titles) - } - }) - t.Run("title not applied", func(t *testing.T) { - current := codex.Task{TaskID: "title", Revision: "1", Title: "Title", Source: "vscode"} - committed := state.New() - committed.LastUpdateCheck = timePointer(now) - committed.Tasks[current.TaskID] = record(current, state.StatusComplete, now) - runner, deps := testRunner(t, now, []codex.Task{current}, committed) - cycle := state.NewCycle("cycle-1", committed.Generation, now) - cycle.Inventory[current.TaskID] = state.CapturedTask{TaskID: current.TaskID, Revision: "1", Title: "Title", LastSubstantiveActivity: now} - cycle.Results[current.TaskID] = state.ClassificationResult{TaskID: current.TaskID, Revision: "1", Status: state.StatusComplete, Provenance: state.ProvenanceFooter, DurableSubject: "Title"} - cycle.Operations["title:title"] = state.CycleOperation{Kind: state.OperationTitle, Stage: state.StageApplying, TaskID: current.TaskID, ExpectedRevision: "1", ExpectedTitle: "Title", DesiredTitle: "✅ Title"} - if err := deps.store.store.SaveCycle(cycle); err != nil { - t.Fatal(err) - } - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) - } - if len(deps.client.titles) != 1 { - t.Fatalf("title was not retried: %v", deps.client.titles) - } - stored, _ := deps.store.store.LoadState() - if stored.Tasks[current.TaskID].CapturedTitle != "✅ Title" || len(stored.PendingTitlePlans) != 0 { - t.Fatalf("state=%+v", stored) - } - }) - t.Run("archive", func(t *testing.T) { - committed := state.New() - committed.LastUpdateCheck = timePointer(now) - runner, deps := testRunner(t, now, nil, committed) - cycle := state.NewCycle("cycle-1", committed.Generation, now) - cycle.Inventory["complete"] = state.CapturedTask{TaskID: "complete", Revision: "1", Title: "✅ Complete", LastSubstantiveActivity: now.Add(-15 * 24 * time.Hour)} - cycle.Results["complete"] = state.ClassificationResult{TaskID: "complete", Revision: "1", Status: state.StatusComplete, Provenance: state.ProvenanceFooter, DurableSubject: "Complete"} - cycle.Operations["archive:complete"] = state.CycleOperation{Kind: state.OperationArchive, Stage: state.StageApplying, TaskID: "complete", ExpectedRevision: "1", ExpectedTitle: "✅ Complete"} - if err := deps.store.store.SaveCycle(cycle); err != nil { - t.Fatal(err) - } - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) - } - stored, _ := deps.store.store.LoadState() - if _, ok := stored.Archives["complete"]; ok || len(deps.client.archives) != 0 { - t.Fatalf("external archive was claimed: archives=%v stored=%+v", deps.client.archives, stored.Archives) - } - }) - t.Run("applied archive", func(t *testing.T) { - committed := state.New() - committed.LastUpdateCheck = timePointer(now) - runner, deps := testRunner(t, now, nil, committed) - cycle := state.NewCycle("cycle-1", committed.Generation, now) - cycle.Inventory["complete"] = state.CapturedTask{TaskID: "complete", Revision: "1", Title: "✅ Complete", LastSubstantiveActivity: now.Add(-15 * 24 * time.Hour)} - cycle.Results["complete"] = state.ClassificationResult{TaskID: "complete", Revision: "1", Status: state.StatusComplete, Provenance: state.ProvenanceFooter, DurableSubject: "Complete"} - cycle.Operations["archive:complete"] = state.CycleOperation{Kind: state.OperationArchive, Stage: state.StageApplied, TaskID: "complete", ExpectedRevision: "1", ExpectedTitle: "✅ Complete"} - if err := deps.store.store.SaveCycle(cycle); err != nil { - t.Fatal(err) - } - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) - } - stored, _ := deps.store.store.LoadState() - if _, ok := stored.Archives["complete"]; !ok || len(deps.client.archives) != 0 { - t.Fatalf("applied archive was not recovered: archives=%v stored=%+v", deps.client.archives, stored.Archives) - } - }) - t.Run("notice", func(t *testing.T) { - committed := state.New() - committed.LastUpdateCheck = timePointer(now.Add(-25 * time.Hour)) - runner, deps := testRunner(t, now, nil, committed) - cfg := config.Default("control") - cfg.AutoUpdateEnabled = false - deps.store.configOverride = &cfg - cycle := state.NewCycle("cycle-1", committed.Generation, now) - cycle.Operations["notice:1.2.0"] = state.CycleOperation{Kind: state.OperationNotice, Stage: state.StageApplying, NoticeVersion: "1.2.0"} - if err := deps.store.store.SaveCycle(cycle); err != nil { - t.Fatal(err) - } - deps.update.result = UpdateStatus{LatestVersion: "1.2.0", Newer: true} - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) - } - stored, _ := deps.store.store.LoadState() - if len(stored.DeliveredNoticeVersions) != 1 || stored.DeliveredNoticeVersions[0] != "1.2.0" || len(deps.client.notices) != 1 || len(deps.client.latestReads) != 0 { - t.Fatalf("delivered=%v notices=%v latestReads=%v", stored.DeliveredNoticeVersions, deps.client.notices, deps.client.latestReads) - } - }) -} - -func TestLegacyCycleRecoveryReconstructsManagedActionAndTokenOwnership(t *testing.T) { - now := time.Date(2026, 7, 26, 11, 0, 0, 0, time.UTC) - const rolloutPath = "/synthetic/legacy-title.jsonl" - expected := codex.Task{TaskID: "title", Revision: "1", Title: "➡️ 1.2m Release service → old action", Source: "vscode", RolloutPath: rolloutPath} - desired := "➡️ Release service → review rollout · out 1.6m" - current := expected - current.Revision = "2" - current.Title = desired + current := codex.Task{TaskID: "title", Revision: "1", Title: "Title", Source: "vscode"} committed := state.New() - committed.BootstrapComplete = true committed.LastUpdateCheck = timePointer(now) - previous := record(expected, state.StatusNextSteps, now.Add(-time.Hour)) - previous.DurableSubject = "Release service" - previous.ManagedAction = "old action" - previous.ManagedTokenDisplay = "1.2m" - previous.ManagedTokenPosition = tokens.PositionStart - previous.TokenDisplayPosition = tokens.PositionStart - previous.TokenRolloutPath = rolloutPath - previous.OutputTokens = 1_200_000 - previous.TokenUsageFound = true - committed.Tasks[current.TaskID] = previous + committed.Tasks[current.TaskID] = record(current, state.StatusComplete, now) runner, deps := testRunner(t, now, []codex.Task{current}, committed) - cfg := config.Default("control") - cfg.TokenDisplay = tokens.PositionEnd - deps.store.configOverride = &cfg - deps.tokens.snapshots[rolloutPath] = tokens.Snapshot{RolloutPath: rolloutPath, Offset: 120, Size: 240, OutputTokens: 1_600_123, TotalTokens: 2_000_000, Found: true} - cycle := state.NewCycle("legacy-title-cycle", committed.Generation, now) - cycle.Inventory[current.TaskID] = state.CapturedTask{TaskID: current.TaskID, Revision: expected.Revision, Title: expected.Title, RolloutPath: rolloutPath, LastSubstantiveActivity: now.Add(-time.Hour)} - cycle.Results[current.TaskID] = state.ClassificationResult{TaskID: current.TaskID, Revision: expected.Revision, Status: state.StatusNextSteps, Provenance: state.ProvenanceFooter, ManagedAction: "review rollout"} - cycle.Operations["title:"+current.TaskID] = state.CycleOperation{Kind: state.OperationTitle, Stage: state.StageApplying, TaskID: current.TaskID, ExpectedRevision: expected.Revision, ExpectedTitle: expected.Title, DesiredTitle: desired} + cycle := state.NewCycle("cycle-1", committed.Generation, now) + cycle.Inventory[current.TaskID] = state.CapturedTask{TaskID: current.TaskID, Revision: "1", Title: "Title", LastSubstantiveActivity: now} + cycle.Results[current.TaskID] = state.ClassificationResult{TaskID: current.TaskID, Revision: "1", Status: state.StatusComplete, Provenance: state.ProvenanceFooter, DurableSubject: "Title"} + cycle.Operations["title:title"] = state.CycleOperation{Kind: state.OperationTitle, Stage: state.StageApplying, TaskID: current.TaskID, ExpectedRevision: "1", ExpectedTitle: "Title", DesiredTitle: "✅ Title"} if err := deps.store.store.SaveCycle(cycle); err != nil { t.Fatal(err) } if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } - stored, err := deps.store.store.LoadState() - if err != nil { - t.Fatal(err) + stored, _ := deps.store.store.LoadState() + if stored.PendingTitlePlans[current.TaskID].DesiredTitle != "✅ Title" || len(deps.client.titles) != 0 { + t.Fatalf("state=%+v writes=%v", stored, deps.client.titles) } - recovered := stored.Tasks[current.TaskID] - live, _ := deps.index.task(current.TaskID) - canonical := "➡️ Release service → review rollout · 1.6m" - if recovered.CapturedRevision != live.Revision || recovered.CapturedTitle != canonical || recovered.LastAppliedTitle != canonical { - t.Fatalf("recovered title ownership = %+v", recovered) +} + +func TestLegacyCycleRecoveryReconstructsManagedActionAndTokenOwnership(t *testing.T) { + now := time.Date(2026, 7, 26, 10, 0, 0, 0, time.UTC) + expected := codex.Task{TaskID: "title", Revision: "1", Title: "➡️ 1.2m Release service → old action", Source: "vscode", RolloutPath: "/synthetic/legacy-title.jsonl"} + committed := state.New() + committed.LastUpdateCheck = timePointer(now) + rec := record(expected, state.StatusNextSteps, now) + rec.DurableSubject = "Release service" + rec.ManagedAction = "old action" + rec.ManagedTokenDisplay = "1.2m" + rec.ManagedTokenPosition = tokens.PositionStart + rec.TokenDisplayPosition = tokens.PositionEnd + committed.Tasks[expected.TaskID] = rec + runner, deps := testRunner(t, now, []codex.Task{expected}, committed) + cycle := state.NewCycle("legacy", committed.Generation, now) + cycle.Inventory[expected.TaskID] = state.CapturedTask{TaskID: expected.TaskID, Revision: expected.Revision, Title: expected.Title, RolloutPath: expected.RolloutPath, LastSubstantiveActivity: now} + cycle.Results[expected.TaskID] = state.ClassificationResult{TaskID: expected.TaskID, Revision: expected.Revision, Status: state.StatusNextSteps, Provenance: state.ProvenanceFooter, ManagedAction: "review rollout"} + cycle.Operations["title:"+expected.TaskID] = state.CycleOperation{Kind: state.OperationTitle, Stage: state.StageApplying, TaskID: expected.TaskID, ExpectedRevision: expected.Revision, ExpectedTitle: expected.Title, DesiredTitle: "➡️ Release service → review rollout"} + if err := deps.store.store.SaveCycle(cycle); err != nil { + t.Fatal(err) } - if recovered.DurableSubject != "Release service" || recovered.ManagedAction != "review rollout" || recovered.ManagedTokenDisplay != "1.6m" || recovered.ManagedTokenPosition != tokens.PositionEnd { - t.Fatalf("recovered semantic ownership = %+v", recovered) + if _, err := runner.Run(context.Background(), false); err != nil { + t.Fatal(err) } - if len(stored.PendingTitlePlans) != 0 || len(deps.client.titles) != 1 { - t.Fatalf("pending=%+v writes=%v", stored.PendingTitlePlans, deps.client.titles) + stored, _ := deps.store.store.LoadState() + plan := stored.PendingTitlePlans[expected.TaskID] + if plan.DesiredTitle != "➡️ Release service → review rollout" || plan.DurableSubject != "Release service" || plan.ManagedAction != "review rollout" || len(deps.client.titles) != 0 { + t.Fatalf("plan=%+v writes=%v", plan, deps.client.titles) } } @@ -2204,85 +2037,67 @@ func migratedPlan(task codex.Task, desired string) state.PendingTitlePlan { func TestPendingTitleMigrationSettlesAlreadyAppliedPlanWithoutReads(t *testing.T) { now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) - expected := codex.Task{TaskID: "task-a", Revision: "1", Title: "Subject", Source: "vscode"} - current := expected - current.Title = "✅ Subject" + task := codex.Task{TaskID: "task-a", Revision: "1", Title: "✅ Subject", Source: "vscode"} committed := state.New() committed.BootstrapComplete = true committed.LastUpdateCheck = timePointer(now) - committed.Tasks[expected.TaskID] = record(expected, state.StatusComplete, now) - plan := migratedPlan(expected, current.Title) - plan.ManagedTokenDisplay = "26k" - plan.ManagedTokenPosition = tokens.PositionEnd - committed.PendingTitlePlans[expected.TaskID] = plan - runner, deps := testRunner(t, now, []codex.Task{current}, committed) - + committed.Tasks[task.TaskID] = record(task, state.StatusComplete, now) + plan := migratedPlan(task, task.Title) + plan.NativeOutcome = state.NativeTitleSucceeded + reported := now.Add(-time.Second) + plan.NativeReportedAt = &reported + plan.CanonicalCheckedAt = &reported + committed.PendingTitlePlans[task.TaskID] = plan + runner, deps := testRunner(t, now, []codex.Task{task}, committed) if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } stored, _ := deps.store.store.LoadState() - owned := stored.Tasks[current.TaskID] - if len(stored.PendingTitlePlans) != 0 || owned.CapturedRevision != current.Revision || owned.CapturedTitle != current.Title || owned.LastAppliedTitle != current.Title || owned.DurableSubject != plan.DurableSubject || owned.ManagedTokenDisplay != plan.ManagedTokenDisplay || owned.ManagedTokenPosition != plan.ManagedTokenPosition { + if len(stored.PendingTitlePlans) != 0 || stored.Tasks[task.TaskID].LastAppliedTitle != task.Title || len(deps.client.titles) != 0 { t.Fatalf("state=%+v", stored) } - if len(deps.client.latestReads) != 0 || len(deps.client.previousReads) != 0 || len(deps.tokens.calls) != 0 || len(deps.client.titles) != 0 || deps.classifier.calls != 0 { - t.Fatalf("latest=%v previous=%v tokens=%v titles=%v classifier=%d", deps.client.latestReads, deps.client.previousReads, deps.tokens.calls, deps.client.titles, deps.classifier.calls) - } } func TestPendingTitleMigrationSettlesAlreadyAppliedAdvancedRevisionWithoutClassifier(t *testing.T) { now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) - expected := codex.Task{TaskID: "task-a", Revision: "1", Title: "Subject", Source: "vscode", RolloutPath: "/synthetic/task-a.jsonl"} - current := expected - current.Revision = "2" - current.Title = "✅ Subject" - evidence := completedEvidence(now, "done", "no footer") + expected := codex.Task{TaskID: "task-a", Revision: "1", Title: "Subject", Source: "vscode"} + current := codex.Task{TaskID: "task-a", Revision: "2", Title: "✅ Subject", Source: "vscode"} committed := state.New() committed.BootstrapComplete = true committed.LastUpdateCheck = timePointer(now) - previous := record(expected, state.StatusComplete, now) - previous.Provenance = state.ProvenanceLuna - previous.EvidenceFingerprint = evidenceFingerprint(expected, evidence) - committed.Tasks[expected.TaskID] = previous + committed.Tasks[expected.TaskID] = record(expected, state.StatusComplete, now) plan := migratedPlan(expected, current.Title) + plan.NativeOutcome = state.NativeTitleSucceeded + reported := now.Add(-time.Second) + plan.NativeReportedAt = &reported + plan.CanonicalCheckedAt = &reported committed.PendingTitlePlans[expected.TaskID] = plan runner, deps := testRunner(t, now, []codex.Task{current}, committed) - deps.client.latest[current.TaskID] = evidence - if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } stored, _ := deps.store.store.LoadState() - owned := stored.Tasks[current.TaskID] - if len(stored.PendingTitlePlans) != 0 || owned.CapturedRevision != current.Revision || owned.CapturedTitle != current.Title || owned.LastAppliedTitle != current.Title || owned.DurableSubject != plan.DurableSubject || owned.Status != previous.Status || owned.Provenance != previous.Provenance { + if len(stored.PendingTitlePlans) != 0 || stored.Tasks[current.TaskID].CapturedRevision != current.Revision || deps.classifier.calls != 0 { t.Fatalf("state=%+v", stored) } - if len(deps.client.latestReads) != 1 || len(deps.client.previousReads) != 0 || len(deps.tokens.calls) != 1 || len(deps.client.titles) != 0 || deps.classifier.calls != 0 { - t.Fatalf("latest=%v previous=%v tokens=%v titles=%v classifier=%d", deps.client.latestReads, deps.client.previousReads, deps.tokens.calls, deps.client.titles, deps.classifier.calls) - } } func TestPendingTitleMigrationPromotesValidPlanWithoutClassificationReads(t *testing.T) { now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) - task := codex.Task{TaskID: "task-a", Revision: "1", Title: "Subject", Source: "vscode", RolloutPath: "/synthetic/task-a.jsonl"} + task := codex.Task{TaskID: "task-a", Revision: "1", Title: "Subject", Source: "vscode"} committed := state.New() committed.BootstrapComplete = true committed.LastUpdateCheck = timePointer(now) committed.Tasks[task.TaskID] = record(task, state.StatusComplete, now) - committed.PendingTitlePlans[task.TaskID] = migratedPlan(task, "✅ Subject") + plan := migratedPlan(task, "✅ Subject") + committed.PendingTitlePlans[task.TaskID] = plan runner, deps := testRunner(t, now, []codex.Task{task}, committed) - if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } - stored, _ := deps.store.store.LoadState() - current, _ := deps.index.task(task.TaskID) - if current.Title != "✅ Subject" || stored.Tasks[task.TaskID].CapturedTitle != current.Title || len(stored.PendingTitlePlans) != 0 || len(deps.client.titles) != 1 { - t.Fatalf("current=%+v state=%+v titles=%v", current, stored, deps.client.titles) - } - if len(deps.client.latestReads) != 0 || len(deps.client.previousReads) != 0 || len(deps.tokens.calls) != 0 || deps.classifier.calls != 0 { - t.Fatalf("latest=%v previous=%v tokens=%v classifier=%d", deps.client.latestReads, deps.client.previousReads, deps.tokens.calls, deps.classifier.calls) + if stored.PendingTitlePlans[task.TaskID].OperationID != plan.OperationID || len(deps.client.titles) != 0 || deps.classifier.calls != 0 { + t.Fatalf("state=%+v", stored) } } @@ -2293,61 +2108,33 @@ func TestPendingTitleMigrationSameTitleForcesOneSetterCall(t *testing.T) { committed.BootstrapComplete = true committed.LastUpdateCheck = timePointer(now) committed.Tasks[task.TaskID] = record(task, state.StatusComplete, now) - committed.PendingTitlePlans[task.TaskID] = migratedPlan(task, task.Title) + plan := migratedPlan(task, task.Title) + committed.PendingTitlePlans[task.TaskID] = plan runner, deps := testRunner(t, now, []codex.Task{task}, committed) - if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } - if len(deps.client.titles) != 1 { - t.Fatalf("titles=%v", deps.client.titles) - } stored, _ := deps.store.store.LoadState() - current, _ := deps.index.task(task.TaskID) - if current.Revision == task.Revision || stored.Tasks[task.TaskID].CapturedRevision != current.Revision || len(stored.PendingTitlePlans) != 0 { - t.Fatalf("current=%+v stored=%+v", current, stored) - } - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) - } - if len(deps.client.titles) != 1 { - t.Fatalf("same-title refresh repeated: %v", deps.client.titles) + if stored.PendingTitlePlans[task.TaskID].OperationID != plan.OperationID || len(deps.client.titles) != 0 { + t.Fatalf("state=%+v writes=%v", stored, deps.client.titles) } } func TestPendingTitleMigrationDropsMissingAndDriftedPlans(t *testing.T) { now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC) - for _, test := range []struct { - name string - inventory []codex.Task - wantReads int - }{ - {name: "missing"}, - {name: "drifted", inventory: []codex.Task{{TaskID: "task-a", Revision: "2", Title: "User edit", Source: "vscode"}}, wantReads: 1}, - } { - t.Run(test.name, func(t *testing.T) { - expected := codex.Task{TaskID: "task-a", Revision: "1", Title: "Subject", Source: "vscode"} - committed := state.New() - committed.BootstrapComplete = true - committed.LastUpdateCheck = timePointer(now) - committed.Tasks[expected.TaskID] = record(expected, state.StatusComplete, now) - committed.PendingTitlePlans[expected.TaskID] = migratedPlan(expected, "✅ Subject") - runner, deps := testRunner(t, now, test.inventory, committed) - if test.wantReads > 0 { - deps.client.latest[expected.TaskID] = completedEvidence(now, "done", "🧵🐻 complete") - } - - if _, err := runner.Run(context.Background(), false); err != nil { - t.Fatal(err) - } - stored, _ := deps.store.store.LoadState() - if len(stored.PendingTitlePlans) != 0 || len(deps.client.latestReads) != test.wantReads { - t.Fatalf("state=%+v titles=%v reads=%v", stored, deps.client.titles, deps.client.latestReads) - } - if test.name == "missing" && len(deps.client.titles) != 0 { - t.Fatalf("missing task title writes=%v", deps.client.titles) - } - }) + expected := codex.Task{TaskID: "task-a", Revision: "1", Title: "Subject", Source: "vscode"} + committed := state.New() + committed.BootstrapComplete = true + committed.LastUpdateCheck = timePointer(now) + committed.Tasks[expected.TaskID] = record(expected, state.StatusComplete, now) + committed.PendingTitlePlans[expected.TaskID] = migratedPlan(expected, "✅ Subject") + runner, deps := testRunner(t, now, nil, committed) + if _, err := runner.Run(context.Background(), false); err != nil { + t.Fatal(err) + } + stored, _ := deps.store.store.LoadState() + if len(stored.PendingTitlePlans) != 0 || len(deps.client.titles) != 0 { + t.Fatalf("state=%+v", stored) } } @@ -2361,15 +2148,13 @@ func TestPendingTitleMigrationRetainsPlanOnFailedDirectWrite(t *testing.T) { plan := migratedPlan(task, "✅ Subject") committed.PendingTitlePlans[task.TaskID] = plan runner, deps := testRunner(t, now, []codex.Task{task}, committed) - deps.client.failTitle[task.TaskID] = true - value, err := runner.Run(context.Background(), false) if err != nil { t.Fatal(err) } stored, _ := deps.store.store.LoadState() - if stored.PendingTitlePlans[task.TaskID].OperationID != plan.OperationID || len(deps.client.titles) != 0 || len(value.(output.HeartbeatResult).Retries) != 1 { - t.Fatalf("state=%+v titles=%v result=%+v", stored, deps.client.titles, value) + if stored.PendingTitlePlans[task.TaskID].OperationID != plan.OperationID || len(deps.client.titles) != 0 || len(value.(output.HeartbeatResult).Retries) != 0 { + t.Fatalf("state=%+v result=%+v", stored, value) } } @@ -2410,36 +2195,14 @@ func TestTitleVerificationIsSavedBeforeSameTaskArchive(t *testing.T) { previous.StateStartedAt = now.Add(-15 * 24 * time.Hour) committed.Tasks[task.TaskID] = previous runner, deps := testRunner(t, now, []codex.Task{task}, committed) - oldActivity := now.Add(-15 * 24 * time.Hour).Unix() - deps.client.latest[task.TaskID] = appserver.RecentEvidence{ThreadStatus: appserver.ThreadStatus{Type: "idle"}, RecencyAt: &oldActivity, Latest: &appserver.EvidenceTurn{ID: "turn", Status: "completed", AgentMessage: "🧵🐻 complete"}} - events := []string{} - deps.index.events = &events - deps.client.events = &events - deps.store.events = &events - + old := now.Add(-15 * 24 * time.Hour).Unix() + deps.client.latest[task.TaskID] = appserver.RecentEvidence{ThreadStatus: appserver.ThreadStatus{Type: "idle"}, RecencyAt: &old, Latest: &appserver.EvidenceTurn{ID: "turn", Status: "completed", AgentMessage: "🧵🐻 complete"}} if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } - joined := strings.Join(events, "|") - ordered := []string{ - "inventory", - "save:title=prepared,archive=prepared", - "inventory", - "save:title=applying,archive=prepared", - "set_title", - "save:title=applied,archive=prepared", - "inventory", - "save:title=verified,archive=prepared", - "inventory", - "archive", - } - position := 0 - for _, event := range ordered { - next := strings.Index(joined[position:], event) - if next < 0 { - t.Fatalf("events=%v missing ordered %q", events, event) - } - position += next + len(event) + stored, _ := deps.store.store.LoadState() + if stored.PendingTitlePlans[task.TaskID].DesiredTitle != "✅ Subject" || len(deps.client.archives) != 0 { + t.Fatalf("state=%+v archives=%v", stored, deps.client.archives) } } @@ -2450,23 +2213,20 @@ func TestCrashApplyingSameTitleRefreshRetriesSetter(t *testing.T) { committed.BootstrapComplete = true committed.LastUpdateCheck = timePointer(now) committed.Tasks[task.TaskID] = record(task, state.StatusComplete, now) - plan := migratedPlan(task, task.Title) - committed.PendingTitlePlans[task.TaskID] = plan runner, deps := testRunner(t, now, []codex.Task{task}, committed) - cycle := state.NewCycle("cycle-applying-refresh", committed.Generation, now) + cycle := state.NewCycle("cycle", committed.Generation, now) cycle.Inventory[task.TaskID] = state.CapturedTask{TaskID: task.TaskID, Revision: task.Revision, Title: task.Title, LastSubstantiveActivity: now} cycle.Results[task.TaskID] = state.ClassificationResult{TaskID: task.TaskID, Revision: task.Revision, Status: state.StatusComplete, Provenance: state.ProvenanceFooter, DurableSubject: "Subject"} cycle.Operations["title:"+task.TaskID] = state.CycleOperation{Kind: state.OperationTitle, Stage: state.StageApplying, TaskID: task.TaskID, ExpectedRevision: task.Revision, ExpectedTitle: task.Title, DesiredTitle: task.Title, DurableSubject: "Subject", ForceWrite: true} if err := deps.store.store.SaveCycle(cycle); err != nil { t.Fatal(err) } - if _, err := runner.Run(context.Background(), false); err != nil { t.Fatal(err) } stored, _ := deps.store.store.LoadState() - if len(deps.client.titles) != 1 || len(stored.PendingTitlePlans) != 0 { - t.Fatalf("titles=%v state=%+v", deps.client.titles, stored) + if stored.PendingTitlePlans[task.TaskID].DesiredTitle != task.Title || len(deps.client.titles) != 0 { + t.Fatalf("state=%+v", stored) } } @@ -2495,3 +2255,59 @@ func TestCrashAppliedTitleRecoversWithoutRepeatingSetter(t *testing.T) { t.Fatalf("titles=%v state=%+v", deps.client.titles, stored.Tasks[current.TaskID]) } } + +func TestSourceNativePlanSettlesOnlyWhenPersistedFooterMatches(t *testing.T) { + now := time.Date(2026, 8, 3, 12, 0, 0, 0, time.UTC) + expected := codex.Task{TaskID: "control", Revision: "1", Title: "✅ Control task", Source: "vscode"} + current := codex.Task{TaskID: "control", Revision: "2", Title: "🙋 Control task → choose the release region", Source: "vscode"} + committed := state.New() + committed.BootstrapComplete = true + committed.LastUpdateCheck = timePointer(now) + committed.Tasks[expected.TaskID] = record(expected, state.StatusComplete, now) + plan := state.PendingTitlePlan{ + OperationID: state.TitleOperationID(expected.TaskID, expected.Revision, expected.Title, current.Title), + TaskID: expected.TaskID, ExpectedRevision: expected.Revision, ExpectedTitle: expected.Title, DesiredTitle: current.Title, + DurableSubject: "Control task", ManagedAction: "choose the release region", NativeOutcome: state.NativeTitleSucceeded, + ExpectedFooter: "🧵🐻 needs input (you): choose the release region", + } + reported := now.Add(-time.Second) + plan.NativeReportedAt = &reported + plan.CanonicalCheckedAt = &reported + committed.PendingTitlePlans[expected.TaskID] = plan + runner, deps := testRunner(t, now, []codex.Task{current}, committed) + deps.store.configOverride = func() *config.Config { cfg := config.Default(expected.TaskID); return &cfg }() + deps.client.latest[current.TaskID] = completedEvidence(now, "choose", "🧵🐻 needs input (you): choose the release region") + if _, err := runner.Run(context.Background(), false); err != nil { + t.Fatal(err) + } + stored, _ := deps.store.store.LoadState() + if len(stored.PendingTitlePlans) != 0 || stored.Tasks[current.TaskID].LastAppliedTitle != current.Title || stored.Tasks[current.TaskID].Status != state.StatusNeedsInput { + t.Fatalf("state=%+v", stored) + } +} + +func TestSourceNativePlanRecomputesWhenPersistedFooterDiffers(t *testing.T) { + now := time.Date(2026, 8, 3, 12, 30, 0, 0, time.UTC) + expected := codex.Task{TaskID: "control", Revision: "1", Title: "✅ Control task", Source: "vscode"} + current := codex.Task{TaskID: "control", Revision: "2", Title: "🙋 Control task → choose the release region", Source: "vscode"} + committed := state.New() + committed.BootstrapComplete = true + committed.LastUpdateCheck = timePointer(now) + committed.Tasks[expected.TaskID] = record(expected, state.StatusComplete, now) + plan := state.PendingTitlePlan{OperationID: state.TitleOperationID(expected.TaskID, expected.Revision, expected.Title, current.Title), TaskID: expected.TaskID, ExpectedRevision: expected.Revision, ExpectedTitle: expected.Title, DesiredTitle: current.Title, DurableSubject: "Control task", ManagedAction: "choose the release region", NativeOutcome: state.NativeTitleSucceeded, ExpectedFooter: "🧵🐻 needs input (you): choose the release region"} + reported := now.Add(-time.Second) + plan.NativeReportedAt = &reported + plan.CanonicalCheckedAt = &reported + committed.PendingTitlePlans[expected.TaskID] = plan + runner, deps := testRunner(t, now, []codex.Task{current}, committed) + deps.store.configOverride = func() *config.Config { cfg := config.Default(expected.TaskID); return &cfg }() + deps.client.latest[current.TaskID] = completedEvidence(now, "done", "🧵🐻 complete") + if _, err := runner.Run(context.Background(), false); err != nil { + t.Fatal(err) + } + stored, _ := deps.store.store.LoadState() + replacement := stored.PendingTitlePlans[current.TaskID] + if replacement.ExpectedFooter != "" || replacement.DesiredTitle != "✅ Control task" || replacement.NativeOutcome != state.NativeTitlePending { + t.Fatalf("replacement=%+v", replacement) + } +} diff --git a/internal/watch/mutations.go b/internal/watch/mutations.go index d437351..bad56ba 100644 --- a/internal/watch/mutations.go +++ b/internal/watch/mutations.go @@ -17,11 +17,27 @@ func findTask(inventory codex.Inventory, taskID string) (codex.Task, bool) { } func revalidate(ctx context.Context, inventory InventoryReader, controlTaskID string, operation state.CycleOperation) (codex.Task, bool, error) { - current, err := inventory.Inventory(ctx, controlTaskID) - if err != nil { - return codex.Task{}, false, err + var task codex.Task + var ok bool + if operation.TaskID == controlTaskID { + reader, available := inventory.(interface { + Task(context.Context, string) (codex.Task, error) + }) + if !available { + return codex.Task{}, false, nil + } + current, err := reader.Task(ctx, controlTaskID) + if err != nil { + return codex.Task{}, false, err + } + task, ok = current, true + } else { + current, err := inventory.Inventory(ctx, controlTaskID) + if err != nil { + return codex.Task{}, false, err + } + task, ok = findTask(current, operation.TaskID) } - task, ok := findTask(current, operation.TaskID) if !ok || task.Archived || task.Revision != operation.ExpectedRevision || task.Title != operation.ExpectedTitle { return codex.Task{}, false, nil } diff --git a/scripts/replay-title-batch.mjs b/scripts/replay-title-batch.mjs new file mode 100644 index 0000000..8b985ad --- /dev/null +++ b/scripts/replay-title-batch.mjs @@ -0,0 +1,57 @@ +import assert from "node:assert/strict"; +import fs from "node:fs"; +import vm from "node:vm"; + +const managed=fs.readFileSync(new URL("../assets/AGENTS.threadbear.md",import.meta.url),"utf8"); +const match=managed.match(/## Native title batch[\s\S]*?```js\n([\s\S]*?)\n```/); +assert(match,"embedded native title batch program is missing"); +const program=match[1]; +assert(!program.includes("process")); +assert(!program.includes("CODEX_THREAD_ID")); +assert(!program.includes("source_task_id")); + +const plans=[ + {operation_id:"op-accepted",task_id:"task-accepted",expected_revision:"1",expected_title:"old-a",desired_title:"new-a"}, + {operation_id:"op-drifted",task_id:"task-drifted",expected_revision:"1",expected_title:"old-b",desired_title:"new-b"}, + {operation_id:"op-failed",task_id:"task-failed",expected_revision:"1",expected_title:"old-c",desired_title:"new-c"}, +]; +const calls=[]; +let listCalls=0; +let rendered=""; +const completed=value=>({output:JSON.stringify(value),exit_code:0}); +const tools={ + exec_command:async({cmd})=>{ + calls.push(cmd); + if(cmd.endsWith("title-batch --json --list")){listCalls++;return listCalls===1?completed({version:1,mode:"list",plans,dispositions:[]}):completed({version:1,mode:"list",plans:[],dispositions:[{operation_id:"op-accepted",task_id:"task-accepted",outcome:"canonical_verified"}]});} + if(cmd.includes("--operation 'op-accepted'")) return completed({version:1,mode:"operation",plans:[plans[0]],dispositions:[]}); + if(cmd.includes("--operation 'op-drifted'")) return completed({version:1,mode:"operation",plans:[],dispositions:[{operation_id:"op-drifted",task_id:"task-drifted",outcome:"drifted"}]}); + if(cmd.includes("--operation 'op-failed'")) return completed({version:1,mode:"operation",plans:[plans[2]],dispositions:[]}); + if(cmd.includes("title-batch --json --report")) return completed({version:1,accepted_ids:["op-accepted"],failed_ids:["op-failed"],drifted_ids:["op-drifted"],rejected_ids:[]}); + throw new Error(`unexpected command: ${cmd}`); + }, + codex_app__set_thread_title:async({threadId,title})=>{ + calls.push(`set:${threadId}:${title}`); + if(threadId==="task-failed") throw new Error("native failure"); + }, +}; +const context=vm.createContext({tools,text:value=>{rendered=value;}}); +assert.equal("process" in context,false); +await vm.runInContext(`(async()=>{${program}\n})()`,context,{timeout:5000}); +assert.deepEqual(JSON.parse(rendered),{accepted_ids:["op-accepted"],canonical_ids:["op-accepted"],failed_ids:["op-failed"],drifted_ids:["op-drifted"],rejected_ids:[]}); +assert.equal(calls[0],"~/.local/bin/threadbear title-batch --json --list"); +assert(calls.indexOf("set:task-accepted:new-a")>calls.findIndex(value=>value.includes("--operation 'op-accepted'"))); +assert(calls.findIndex(value=>value.includes("title-batch --json --report"))>calls.indexOf("set:task-accepted:new-a")); +for(const privateValue of ["old-a","new-a","old-b","new-b","old-c","new-c","task-accepted","task-drifted","task-failed"]){assert(!rendered.includes(privateValue));} +console.log("fresh V8 title batch replay passed"); + +let malformedSetterCalls=0; +const malformedTools={ + exec_command:async({cmd})=>{ + if(cmd.endsWith("title-batch --json --list")) return completed({version:1,mode:"list",plans:[{operation_id:"bad"}],dispositions:[]}); + throw new Error(`unexpected malformed command: ${cmd}`); + }, + codex_app__set_thread_title:async()=>{malformedSetterCalls++;}, +}; +const malformedContext=vm.createContext({tools:malformedTools,text:()=>{}}); +await assert.rejects(vm.runInContext(`(async()=>{${program}\n})()`,malformedContext,{timeout:5000}),/invalid_title_operation/); +assert.equal(malformedSetterCalls,0); diff --git a/site/install b/site/install index 091ba12..f48fc4b 100644 --- a/site/install +++ b/site/install @@ -196,6 +196,12 @@ verifies the published checksum, candidate health, and embedded version before delegating to the candidate. The official-download promise belongs in the opening turn above; do not repeat it as a separate progress message. +## Native title convergence during guided install + +The person's installation consent enables only the managed title-batch protocol in the retained control task. It is not recurring consent for other native operations and must not be requested again on later control-task turns. After the bounded post-install handoff, stage the exact terminal footer, then run the fixed native title-batch program embedded in the installed ThreadBear AGENTS block inside one `functions.exec`. Drain every deterministic operation staged within the handoff window, but do not wait for unresolved semantic classification. Revalidate every operation immediately before the native setter and return only aggregate accepted, failed, drifted, and rejected operation IDs. + +A successful install close may claim visible title convergence only for accepted operations that ThreadBear subsequently verifies canonically. Missing native capability, failed mutation, identity disagreement, drift, or unverified canonical persistence is an explicit partial/failure result. Never expose title or manifest content to the model, create a child task, use the persistent task for unrelated work, drive the UI, restart Codex, or use private IPC/cache/database mutation. + ## 3. Identify this task backstage Before downloading, installing, renaming, pinning, or changing any managed resource, feature-detect the available Codex task tooling: @@ -816,10 +822,13 @@ outcomes below. When it is retryable, keep the healthy-install sentence, say plainly that ThreadBear will keep working on the aggregate retry count, and do not claim convergence. -The background heartbeat performs managed title writes directly through the -pinned App Server. Each title mutation is revision/title revalidated, journaled as applying and applied, verified through a fresh inventory read, journaled as -verified, and only then followed by any same-task archive. Do not run a separate -title worker, helper manifest, native report, child model, or replay step. +The background heartbeat deterministically stages exact, revision/title-guarded +title operations and never treats detached App Server persistence as Desktop +repaint proof. The retained, opted-in control task drains staged operations +through the fixed native title-batch program, reports only aggregate operation +IDs, and leaves canonical verification and commit to ThreadBear. A same-task +archive waits for accepted native application and canonical settlement. Do not +create a title child, recurring hosted automation, or control-task relay. Feature-detect and fail closed when required App Server methods or the tool-free classifier boundary are absent. The ordinary App Server stays on the real Codex