diff --git a/BUILD-LOG.md b/BUILD-LOG.md index 84c08f78a..94cc84039 100644 --- a/BUILD-LOG.md +++ b/BUILD-LOG.md @@ -6013,3 +6013,4 @@ Owner directed static-only fallback after actual nativepriority drift: no new pr 2026-11-04T03:48:47Z | GPT-6.1 Sol implementation sub-agent, t3/pm-intake-tools | #797 / Gitea #1839 lifecycle Task 6a before/after | Assigned base 6dd4f92b0; harness had no intake tools and staff.request required a task. Added intake_open/request.open and intake_propose/request.propose only for PM holding message.send, plus intake_status/request.get and intake_projects/request.projects in PM and CTO pre-task manifests, independent of tracker/checkouts. Existing broker authority is unchanged: its request reads still refuse CTO (request-access-refused); tested and retained rather than widening authority. intake_question omitted because no backing request.question verb exists. staff.request/request_staff takes exactly one of request or task_ref with roles/note/decision unchanged, PM-only message.send and chief routing preserved; original request provenance/reachability uses the existing host check without a fabricated task or opened journal. Exported the existing runner toolServer for acceptance tests against production dispatch, including MCP; no new imports or sandbox/process mounts needed (import closure 1/1). PM duty prose and the generated PM staffing prompt describe pre-task intake and staffing. New tests cover actual PM/CTO/coder/reviewer/assistant bundles for pi and claude-code, assistant -> chief -> PM relays, exact user-mode decision with null task/requirement, duplicate open, wrong roles, invalid/unreachable input, exclusive staff references and assistant coordination restrictions. RED on assigned parent: 6 pass/12 fail (dispatch not exported); after exporting only the existing dispatch seam 7 pass/11 fail for missing tools/staff intake. First GREEN attempt 17/18 exposed a test fixture treating message.receive's array as an object, corrected without weakening assertions. Existing required tests first 53/55 and full harness first 222/223 failed only outdated manifest/schema/prompt expectations introduced by this slice; original parent reruns required 55/55 and bundle 22/22 passed. A further generated staffing prompt RED 21/22 proved the stale task-only instruction before updating it. Final gates: request-tools 18/18, required command 55/55, full harness 223/223, bus 1477/1477, business 305/305, sandbox import closure 1/1, all zero failures/skips/cancellations. Full logs, baseline extraction and red evidence: /mnt/storage/scratch/tmp/JHTDYm/; every run used that TMPDIR with pipefail. Local commit only; no push, no role JSON/contract/migration/schema/config/credential changes, no runtime/bootstrap execution, no continuation/PRD/advance/recovery work. Controller owns review, integration and queue/tracker coordination; linked-worktree queue command refused as designed. 2026-11-04T04:06:34Z | GPT-6.1 Sol implementation sub-agent, t3/lifecycle-task-5a-create-recovery | #797 / Gitea #1839 lifecycle Task 5a before/after | Before: 6dd4f92b0 rejects operation keys and cannot recover a lost tracker create response. After: optional op reserves an immutable action.allowed tracker.create.prepared event, complete POST body/digest, target board and original PM/bot identity before any create-path network I/O; optional project_key and requirement_snapshot are opaque evidence and visible marker fields. Reusing an op cannot POST again, even after a complete negative scan or a restart before POST. Sync-only id-keyed full scans include closed tasks and require one marker, unchanged create fields, original board and creator; edited/missing/duplicate markers and partial/failed scans retain a deduplicated action.refused tracker.create.wait reconciliation receipt. Host-only recordRecoveredCreate validates the prepared receipt, body, creator and board bucket, atomically records the missing self snapshot/task.created once with original attribution, and has no socket route. Adapter exposes reconcileCreates and inspectProject, startup recovers; live PM retries resume labels/relations through read-before-write and current authority checks, assignment uses the existing separate verb. Legacy calls without new fields preserve their existing behavior; lifecycle callers must reuse a stable op for safe retries. Scope-only calls receive a generated op recorded in evidence. No migration, schema-pin edits, role/contract changes or runtime.mjs edits; packages/tasks/requirements.json and its consistency test unchanged. RED: new recovery suite 0/16 on parent; final regressions 0/21 on parent, expected invalid-request for op. Authority and forged-bucket checks also observed RED before their fixes. Corrections: event kind/subject guards required existing action event kinds instead of inventing task/ tracker kinds; restart fixtures needed historical launch cleanup before a successor claim; initial npm workspace invocation refused before tests and was replaced with per-package npm test. First full bus 1473/1477 exposed the added broker import outside the sandbox source closure, including three managed-launch failures; moved pure identity helpers into already-mounted broker.mjs, with parent project-runtime 3/3 and model-hire-ownership 18/18, then focused 42/42. Second bus 1476/1477 encountered unchanged row 280 sshd fixture cleanup ENOTEMPTY and hung; terminated only that run's owned host-run test and sshd, then waited for completion. Parent host-run was 25/26 with T27 failure (row 280 passed); local isolated host-run 26/26. Final GREEN: recovery 21/21, required verbs/adapter/tree/requirements 61/61, full tasks npm test 121/121, full bus npm test 1477/1477, zero failures/skips, diff check clean. Full logs and parent archive: /mnt/storage/scratch/tmp/CkjqAm/ (gate-recovery.log, gate-required.log, gate-tasks.log, gate-bus.log plus diagnostic history). Queue command refused this linked worktree, so no queue mutation; lead owns integration/queue. Task 5b catalogue validation, requirementsFor, built-in binding, task scope filtering and push product binding remain deferred as directed, pending Task 4. One local commit only; no push or live tracker/network/instance changes. 2026-11-04T04:09:35Z | Codex implementation sub-agent, t3/lifecycle-prd-approval-activation | #797 / Gitea #1839 lifecycle Task 4 core | Before: parent 6dd4f92b0 has intake/bootstrap and schema 3p, with no exact product PRD activation authority. After: bounded committed candidate reads (64 requirements, 32 KiB PRD, 16 KiB requirement text, 256 KiB aggregate), strict records, server-derived blob IDs, selected-path HEAD equality and current owner spec; PM-only record.write submission raises the existing nonblocking prd.approve action. Autonomous routing requires a distinct authorized technical arbiter instance and definition; unusable/self routing records setup waits. Guided approval uses the same action without a routing citation, while retaining its spec/mode binding. Host-only catalog activation rereads candidate/spec/mode/current role authority and the exact approving resolution, independent of run liveness; snapshot, activation event, continuation and receipt share one SQLite transaction, with deterministic idempotent replay. Broker.requirementsFor keeps the agreed authenticated consumer signature; no tasks are manufactured. Mode/spec changes invalidate admissions and revoke snapshots; owner-only prd.revoke.console is revision/op checked and replayable, and request.draft.console reads only the bound committed candidate. Runtime recovers from journal/records alone and drains queued lifecycle work on close. New append-only migration 3p to 3q widens request event kinds, preserving rows, sequences and all guards; current version/fixture pins updated, table/guard counts stay 30/114. RED on 6dd4f92b0 with the final new tests: 0/27 pass, 27 expected missing-feature failures (baseline-new-tests-red.log). Corrections: the first fixture tried to initialize an already bootstrapped records repo, fixed before feature RED; journal composition initially attempted nested transactions, now composed inside one host transaction; new automatic closure continuation needed shutdown draining; one revoke assertion needed awaited rejection. First full bus run 1496/1498: the old two-verb whitelist pin was updated to the authorized lifecycle verbs, and unchanged #656 FIFO child measured niceness 16 rather than 19. Both files pass 38/38 on exact parent 6dd4f92b0 and on this branch; the FIFO failure did not reproduce in isolation. Final GREEN: new suites 27/27, requested kept-green trio 21/21, bus 1498/1498, project 50/50, business 305/305, all zero skips; import-closure checks 3/3, git diff --check clean. Full logs under /mnt/storage/scratch/tmp/wZ4am1 (focused-final.log, kept-green-final.log, bus-final.log, project.log, business.log; earlier runs retained). Boundaries: Console HTTP/browser files, revoke route and request-panel UI deferred until Task 3 integration; tracker/task consumer wiring belongs to Task 5; built-in legacy binding is deliberately unset here. The Stack PRD amendment remains proposed. Queue command refused this linked worktree, so queue movement stays with the canonical coordinator. One local commit authorized, no push, merge, deployment, live instance changes, role/contract edits or secret access. +2026-11-04T17:21:41Z | Codex implementation sub-agent, t3/lifecycle-task-6b | #797 / Gitea #1839 lifecycle Task 6b before/after | Before: origin/refactor 103168614 includes Task 6a intake/staff tools and Task 5a create recovery; Task 4 source 76207fe48 cherry-picked as e41ec8159d23d52963d869abf22fc9d430831f44 with both append-only log sides retained. No checked task/review advancement or durable role continuation recovery existed. After: PM-only submit_prd and advance_request map to exact PRD submission and evidence-derived planning → implementation → review → delivery → complete transitions (changes return to implementation); task-create/assignment, latest project push/review/verdict and closure evidence are rechecked under the request revision and current snapshot. Eligible PM/CTO intake_status path reads expose only selected committed draft documents. Durable continuation intents are keyed by request/revision/target/step and composed atomically with outgoing messages; immutable attempt, manifest and exact launch-refusal receipts deduplicate retries. Message reads never acknowledge progress; validated lifecycle transitions and the chief's delivery effect in the originating assistant session do. Startup/event recovery plus a single-flight 30-second pass bounded to 64 requests catches missed wakeups. Successors receive the same unfinished key, with existing routed/effect receipts rechecked before resending. The launcher starts only the configured chief through its existing launchPm path; workers use that chief's real launch/end authority. Old product manifests drain their current turn, then the chief relaunches with request.admission and real ensureCheckouts. Exact refusals appear as RequestState.launch_refusal for intake-panel integration; no Console panel is present in this base. No synthetic capability, PM launch grant, hire through the empty chief hireable list, roles/*.json change, extra scheduler service, live-instance action, configuration/credential mount or new worker import was added. Six role duty documents updated. Migration 3q-3r is necessary solely to add the checked advanced lifecycle event to the closed request_events kind list; it preserves rows, sequence and all append-only guards, with historical fixture/version pins updated. Corrections: migration digest/pins and one bundle tool-list assertion were updated; notification scheduling was narrowed so reads do not cause unsolicited bootstrap retries; an accidental repository-root npm test invocation reported no test script and was corrected to packages/harness. One broad bus run hit the pre-existing #656 FIFO child niceness 16 vs 19 assertion (also recorded by Task 4); that test passed 21/21 in isolation and the final full bus rerun passed without a source change to it. Final GREEN, zero skips: new 43/43, focused compatibility 56/56, harness 227/227, bus 1512/1512, business 305/305, launcher 36/36; extra sandbox import closure 11/11, git diff --check clean. Logs: /mnt/storage/scratch/tmp/task6b-final-new.log, task6b-final-focused.log, task6b-final-harness.log, task6b-final-bus.log, task6b-final-business.log, task6b-final-launcher.log and task6b-final-sandbox.log in the same directory; earlier failure/isolated logs retained. All test invocations used fresh TMPDIR directories from /mnt/storage/scratch/tmp/XXXXXX. Queue command refused this linked worktree, so no queue movement was attempted in the canonical checkout. One local implementation commit only; no push, merge, release, deployment or live-instance changes. diff --git a/docs/SESSIONS.md b/docs/SESSIONS.md index fc9a7b54e..4b16271bb 100644 --- a/docs/SESSIONS.md +++ b/docs/SESSIONS.md @@ -2002,3 +2002,4 @@ are never rewritten or removed; corrections are new entries. 2026-11-03 | GPT-6.1 Sol implementation sub-agent, t3/lifecycle-task-5a-create-recovery | #797 / Gitea #1839 lifecycle Task 5a | Optional journaled create operations, sync-only complete-scan adoption, host-only recovery with original attribution and PM follow-up resume; no migration. Recovery 21/21, required 61/61, tasks 121/121, bus 1477/1477, zero skips; diagnostics and parent reruns in /mnt/storage/scratch/tmp/CkjqAm/. Task 5b deferred, queue left to lead after linked-worktree refusal; one local commit only, no push or live changes. 2026-11-04 | Codex implementation sub-agent, t3/lifecycle-prd-approval-activation | #797 / Gitea #1839 lifecycle Task 4 core | Exact committed product PRD approval/activation, authenticated requirementsFor, owner revoke/mode/draft broker controls, atomic append-only snapshots/receipts and shutdown-safe recovery; migration 3p to 3q. New 27/27, kept-green 21/21, bus 1498/1498, project 50/50, business 305/305, zero skips; logs /mnt/storage/scratch/tmp/wZ4am1. Console HTTP/browser slice deferred for Task 3 integration. Local commit only, no push. +2026-11-04 | Codex implementation sub-agent, t3/lifecycle-task-6b | #797 / Gitea #1839 Task 6b | Typed PRD/advance and scoped draft reads; durable continuations, evidence acknowledgements, chief launch refusal/retry and manifest drain/relaunch; migration 3q-3r. New 43/43, focused 56/56, harness 227/227, bus 1512/1512, business 305/305, launcher 36/36, extra sandbox 11/11, zero skips; /mnt/storage/scratch/tmp/task6b-final-*.log. Task 4 cherry-pick e41ec8159 from 76207fe48; one local commit only, no push/live changes, queue left to canonical coordinator after linked-worktree refusal. diff --git a/packages/bus/schema.sql b/packages/bus/schema.sql index bc4878401..e90fe024e 100644 --- a/packages/bus/schema.sql +++ b/packages/bus/schema.sql @@ -535,7 +535,7 @@ CREATE TABLE request_events ( business TEXT NOT NULL, request TEXT NOT NULL REFERENCES events(id), revision INTEGER NOT NULL CHECK (revision >= 1), - kind TEXT NOT NULL CHECK (kind IN ('opened','selection.proposed','selection.staged','selection.confirmed','setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','cancelled')), + kind TEXT NOT NULL CHECK (kind IN ('opened','selection.proposed','selection.staged','selection.confirmed','setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','advanced','cancelled')), actor_role TEXT, actor_run TEXT, at TEXT NOT NULL, body TEXT NOT NULL CHECK (json_valid(body) AND json_type(body) = 'object'), diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs index 08d64c6db..85923e44f 100644 --- a/packages/bus/src/broker.mjs +++ b/packages/bus/src/broker.mjs @@ -431,6 +431,7 @@ function checkBusiness(name, b) { export class Broker { #store; #requests; + #requestRecoveryPending = false; #businesses; #sessions = new Map(); #runs = new Map(); @@ -589,6 +590,20 @@ export class Broker { if (typeof requestLifecycle !== 'function') throw new TypeError('requestLifecycle must be a host factory'); this.#requests = requestLifecycle({ dataRoot, + progress: state => this.#requestProgress(state), + continuationContext: (state, target) => this.#continuationContext(state, target), + continuationEffect: (state, intent, message) => { + if (intent.step !== 'send_message' || state.phase !== 'complete' || !state.thread) return null; + const b = this.#business(state.business); + const rows = this.#store.all('SELECT m.id,m.to_role FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE m.business=? AND m.from_role=? AND mt.thread=? AND mt.session=? AND m.in_reply_to=? AND m.seq>(SELECT seq FROM messages WHERE id=?) ORDER BY m.seq', state.business, intent.target, state.thread, state.session, state.source_message, message); + const sent = rows.find(m => b.roles[m.to_role]?.definition === 'assistant'); + return sent ? { message: sent.id, to: sent.to_role } : null; + }, + continuationRouted: (message, run) => !!this.#store.get("SELECT 1 FROM deliveries WHERE message=? AND holder_run=? AND op IN ('routed','delivered')", message, run), + continuationRetry: (state, message, target, context) => { + if (context.chief && target !== context.chief) this.#notice(state.business, message, context.chief, target, NOTICE_ROLE, this.#business(state.business).launch.instances.includes(target), this.#store.get('SELECT body FROM messages WHERE id=?', message)?.body); + }, + continuationMessage: (state, target, body, context) => this.#continuationMessage(state, target, body, context), origin: (identity, request) => this.#requestOrigin(identity, request), authorize: (identity, state, operation) => this.#requestAuthorize(identity, state, operation), project: (business, project) => { @@ -619,6 +634,106 @@ export class Broker { throw new TypeError('invalid request lifecycle'); } } + #scheduleRequests() { + if (!this.#requests || !this.#requests.hasWork?.() || this.#requestRecoveryPending) return; + this.#requestRecoveryPending = true; + setImmediate(() => { + this.#requestRecoveryPending = false; + Promise.resolve().then(() => this.#requests.recover({ limit: 64 })).catch(() => {}); + }); + } + #continuationContext(state, target) { + const b = this.#business(state.business), holder = this.#holder(state.business, target); + const chief = b.launch?.by ?? null, h = chief ? this.#holder(state.business, chief) : null; + const run = holder?.op === 'claim' ? holder.holder_run : null; + const revision = this.#store.get("SELECT max(seq) n FROM events WHERE business=? AND kind IN ('session.launched','session.ended','launch.revoked','launch.restored')", state.business).n; + const claims = this.#store.get('SELECT max(seq) n FROM role_claims WHERE business=?', state.business).n; + const revoked = this.#store.get('SELECT state FROM launch_state WHERE business=?', state.business)?.state === 'revoked'; + return { run, chief, chief_run: h?.op === 'claim' ? h.holder_run : null, revoked, + prerequisite: digestPrd({ business: b, revision, claims, revoked, credentials: this.#credentialStatus() }) }; + } + #continuationMessage(state, target, body, context) { + this.#processingWork(); + const b = this.#business(state.business); + if (!b.roles[target]) fail('unknown-role'); + this.#secretCheck(body); + const mid = randomUUID(); + this.#store.run('INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES(?,?,?,?,?,?,?,?)', + mid, this.#now(), state.business, NOTICE_ROLE, NOTICE_RUN, target, 'RESULT', body); + this.#route(state.business, mid, target, null); + if (!context.run && context.chief && target !== context.chief) + this.#notice(state.business, mid, context.chief, target, NOTICE_ROLE, b.launch.instances.includes(target), body); + return mid; + } + // Host-only. The caller is still the real chief capability, checked for its current launch + // grant. Release only after the runner reports its turn finished (message.read), never as an ack. + requestChief(business) { + const b = this.#business(identifier(business)); + this.#processingWork(); + if (!b.launch || b.roles[b.launch.by]?.definition !== 'chief-of-staff') fail('launch-not-configured'); + if (this.#store.get('SELECT state FROM launch_state WHERE business=?', business)?.state === 'revoked') fail('launch-revoked'); + if (this.#firing(business, b.launch.by)) fail('instance-firing'); + return { instance: b.launch.by }; + } + requestRelease(cap, record) { + const s = this.#session(cap); + keys(record, ['business','run','instance','project'], ['business','run','instance','project']); + if (s.business !== record.business || s.role !== this.#business(s.business).launch?.by || s.role === record.instance) fail('not-authorized'); + if (!this.#business(s.business).launch.instances.includes(record.instance)) fail('instance-not-listed'); + this.#checkAuthority(s, 'role.launch', { target: record.instance }); + this.#checkAuthority(s, 'role.end', { target: record.instance }); + const session = this.#runs.get(record.business + ':' + record.run) ?? fail('unknown-run'); + if (session.role !== record.instance || session.thread) fail('not-authorized'); + if (!this.#requests.launches(record.business).some(i => i.target === record.instance && i.project === record.project)) fail('request-not-found'); + const holder = this.#holder(record.business, record.instance); + if (holder?.op !== 'claim' || holder.holder_run !== record.run) fail('not-holder'); + if (!this.#received.has(record.business + ':' + record.run)) return { released: false }; + const { unread } = this.#store.get("SELECT count(*) unread FROM deliveries d JOIN messages m ON m.id=d.message WHERE m.business=? AND d.holder_run=? AND d.op='delivered' AND NOT EXISTS(SELECT 1 FROM deliveries r WHERE r.message=m.id AND r.op='read')", record.business, record.run); + if (unread) return { released: false }; + this.#store.transaction(() => this.#claimEnd(session, holder, 'release', null, null, 'request-relaunch')); + this.#launcherReleased.add(record.business + ':' + record.run); + return { released: true }; + } + #requestProgress(state) { + const b = this.#business(state.business), staff = state.selection?.bindings.find(x => x.digest === state.plan_digest)?.plan.staff; + if (!staff || !state.snapshot) fail('request-evidence-required'); + const events = this.#store.all('SELECT * FROM events WHERE business=? ORDER BY seq', state.business).map(e => ({ ...e, body: JSON.parse(e.body) })); + const created = events.filter(e => e.kind === 'task.created' && e.actor_role === state.pm && e.body.request === state.request && + e.body.project_key === state.project && e.body.requirement_snapshot === state.snapshot && + events.some(p => p.id === e.body.create_receipt && p.body.operation === 'tracker.create.prepared')); + if (!created.length || state.candidate.requirements.some(r => !created.some(e => e.body.requirement === r.id))) fail('request-evidence-required'); + const tasks = [...new Set(created.map(e => e.subject))].sort(), evidence = created.map(e => e.id); + if (state.phase !== 'planning' && digestPrd(tasks) !== digestPrd([...state.tasks].sort())) fail('request-evidence-required'); + let changes = false; + for (const task of tasks) { + const current = this.#store.get('SELECT fields FROM task_current WHERE business=? AND task_ref=?', state.business, task); + const fields = current && JSON.parse(current.fields); + if (!fields || fields.gone) fail('request-evidence-required'); + const history = events.filter(e => e.subject === task); + const assigned = history.findLast(e => e.kind === 'task.assigned'); + if (!assigned || assigned.actor_role !== state.pm || assigned.body.role !== staff.coder || !fields.assignees?.includes(b.roles[staff.coder]?.tracker?.botId)) fail('request-evidence-required'); + evidence.push(assigned.id); + if (state.phase === 'planning') continue; + const push = history.findLast(e => e.kind === 'action.allowed' && e.body.action === 'git.push.working'); + const review = push && history.findLast(e => e.kind === 'review.requested' && e.seq > push.seq && e.body.commit === push.body.commit && e.body.project === state.project && e.body.to === staff.reviewer); + if (!push || push.actor_role !== staff.coder || push.body.project !== state.project || !review) fail('request-evidence-required'); + evidence.push(push.id, review.id); + if (state.phase === 'implementation') continue; + const verdict = history.findLast(e => e.kind === 'review.verdict' && e.body.request === review.id && e.body.commit === push.body.commit && e.actor_role === staff.reviewer); + if (!verdict) fail('request-evidence-required'); + evidence.push(verdict.id); + if (verdict.body.verdict === 'changes') changes = true; + else if (verdict.body.verdict !== 'approve') fail('request-evidence-required'); + if (state.phase === 'delivery') { + const close = history.findLast(e => e.kind === 'task.closed' && e.body.verdict === verdict.id && e.seq > verdict.seq); + if (changes || !close || close.actor_role !== state.pm || fields.done !== true) fail('request-evidence-required'); + evidence.push(close.id); + } + } + const phase = { planning: 'implementation', implementation: 'review', review: changes ? 'implementation' : 'delivery', delivery: 'complete' }[state.phase]; + if (!phase) fail('request-evidence-required'); + return { phase, tasks, evidence: [...new Set(evidence)] }; + } #requestAuthorize(s, state, operation) { if (!s || s.business !== state.business) fail('request-access-refused'); if (s.reader) fail('read-only'); @@ -628,10 +743,12 @@ export class Broker { if (s.human) { if (!s.console) this.#human(s); if (s.human !== state.owner || s.human !== this.#business(s.business).human) fail('request-access-refused'); - if (['propose','prd.submit'].includes(operation)) fail('agent-required'); + if (['propose','prd.submit','advance'].includes(operation)) fail('agent-required'); } else { this.#agent(s); - if (s.role !== state.pm || this.#business(s.business).roles[s.role]?.definition !== 'pm') fail('request-access-refused'); + const b = this.#business(s.business); + const technical = operation === 'get' && b.arbiters?.technical === s.role && b.roles[s.role]?.definition === 'cto'; + if (!technical && (s.role !== state.pm || b.roles[s.role]?.definition !== 'pm')) fail('request-access-refused'); // No separate authority: intake preparation consumes the PM's existing message.send grant. if (operation === 'prd.submit') this.#checkAuthority(s, 'record.write', { target: state.project }); else if (operation !== 'get') this.#checkAuthority(s, 'message.send', { target: state.pm }); @@ -720,6 +837,7 @@ export class Broker { checkBusiness(name, next); if (next.human !== this.#businesses[name].human) fail('invalid-business'); this.#businesses[name] = next; + this.#scheduleRequests(); } // Lifecycle Task 2 (#797): the chat project a bootstrapped product lists under. Host-only, so no verb // reaches it. State is 'free', 'associated' (this exact id, name and product) or 'conflict' (anything else @@ -1743,6 +1861,8 @@ export class Broker { subject, JSON.stringify(body), ); + if (['session.launched','session.ended','task.created','task.assigned','task.closed','review.requested','review.verdict','launch.revoked','launch.restored'].includes(kind) || + (kind === 'action.allowed' && ['git.push.working','record.write','processing.resume'].includes(body?.action))) this.#scheduleRequests(); return eid; } // Row 234 (#1740): grant evidence, through writeGrantEvent. Caller owns the transaction. @@ -5137,13 +5257,20 @@ export class Broker { ? this.#requests.catalog.revoke(s, structuredClone(args)) : this.#requests.modeConsole(s, structuredClone(args))) .catch(e => { throw this.#refused(s, new BusError(e.code ?? 'request-failed'), request); }); } - if (['request.open','request.get','request.projects','request.propose','prd.submit'].includes(request.verb)) { + if (request.verb === 'requirements' && Object.keys(args).length) { + if (s.reader) fail('read-only'); + if (s.tester) fail('tester-refused'); + keys(args, ['project_key','requirement_snapshot','requirement'], ['project_key']); + this.#agent(s); + return this.requirementsFor(cap, structuredClone(args)); + } + if (['request.open','request.get','request.projects','request.propose','prd.submit','request.advance'].includes(request.verb)) { if (s.reader) fail('read-only'); if (s.tester) fail('tester-refused'); if (!this.#requests) fail('request-not-configured'); const method = request.verb === 'prd.submit' ? 'submit' : request.verb.slice('request.'.length); return Promise.resolve().then(() => this.#requests[method](s, structuredClone(args))) - .then(result => { this.#secretCheck(result); return result; }) + .then(result => { this.#secretCheck(result); if (request.verb === 'prd.submit') this.#scheduleRequests(); return result; }) .catch(e => { throw this.#refused(s, e instanceof BusError ? e : new BusError(e.code ?? 'request-failed'), request); }); } // Row 213 (C2-08, Case 2 design 5.1 M6): a creative role has no free-text decision channel. It @@ -5279,6 +5406,7 @@ export class Broker { // branches checks `reader`: each passes a tester by `s.tester`, and each write records `tester`. const view = s.tester && !request.verb.startsWith('message.') ? { ...s, reader: true } : s.user ? { ...s, grants } : s; const result = this.#dispatch(view, request.verb, args); + if (['role.claim','role.release','message.send'].includes(request.verb)) this.#scheduleRequests(); this.#secretCheck(result); return result; }); @@ -7615,7 +7743,7 @@ export class Broker { // One notice per waiting message: only message.send calls this, once for the message it stores, so // a re-route at a claim or a broker restart never sends another. Its class is RESULT: the runner // takes a turn on it and sends no automatic reply, which NOTICE_ROLE could not receive. - #notice(business, mid, by, to, sender, launchable) { + #notice(business, mid, by, to, sender, launchable, continuation = null) { const nid = randomUUID(); this.#store.run( 'INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,in_reply_to,decision,corrects,body) VALUES(?,?,?,?,?,?,?,?,?,?,?)', @@ -7629,9 +7757,9 @@ export class Broker { mid, null, null, - launchable + (launchable ? `Message ${mid} from ${sender} to ${to} waits: ${to} has no running session. Launch ${to} within the launch block's limits, or tell ${sender} why not with message.send in_reply_to ${mid}.` - : `Message ${mid} from ${sender} to ${to} waits: ${to} has no running session and isn't in the launch block's instances, so a launch of ${to} is refused. Tell ${sender} why not with message.send in_reply_to ${mid}.`, + : `Message ${mid} from ${sender} to ${to} waits: ${to} has no running session and isn't in the launch block's instances, so a launch of ${to} is refused. Tell ${sender} why not with message.send in_reply_to ${mid}.`) + (continuation ? `\nWorker evidence (resume after launch):\n${continuation}` : ''), ); this.#route(business, nid, by, null); } diff --git a/packages/bus/src/migrations/3q-3r.sql b/packages/bus/src/migrations/3q-3r.sql new file mode 100644 index 000000000..d2425b276 --- /dev/null +++ b/packages/bus/src/migrations/3q-3r.sql @@ -0,0 +1,45 @@ +-- Task 6b: preserve all evidence and guards, widen only lifecycle event kinds. +DROP TRIGGER request_events_origin; +DROP TRIGGER request_events_revision; +DROP TRIGGER request_effects_binding; +DROP TRIGGER request_events_no_update; +DROP TRIGGER request_events_no_delete; +DROP TRIGGER request_events_no_replace; +DROP INDEX request_events_request; +ALTER TABLE request_events RENAME TO request_events_3q; +CREATE TABLE request_events ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + business TEXT NOT NULL, + request TEXT NOT NULL REFERENCES events(id), + revision INTEGER NOT NULL CHECK (revision >= 1), + kind TEXT NOT NULL CHECK (kind IN ('opened','selection.proposed','selection.staged','selection.confirmed','setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','advanced','cancelled')), + actor_role TEXT, actor_run TEXT, + at TEXT NOT NULL, + body TEXT NOT NULL CHECK (json_valid(body) AND json_type(body) = 'object'), + CHECK ((actor_role IS NULL) = (actor_run IS NULL)), + UNIQUE (business,request,revision) +) STRICT; +INSERT INTO request_events(seq,business,request,revision,kind,actor_role,actor_run,at,body) + SELECT seq,business,request,revision,kind,actor_role,actor_run,at,body FROM request_events_3q ORDER BY seq; +DELETE FROM sqlite_sequence WHERE name='request_events'; +INSERT INTO sqlite_sequence(name,seq) SELECT 'request_events', seq FROM sqlite_sequence WHERE name='request_events_3q'; +DROP TABLE request_events_3q; +CREATE INDEX request_events_request ON request_events (business,request,seq); +CREATE TRIGGER request_events_origin BEFORE INSERT ON request_events + WHEN NOT EXISTS (SELECT 1 FROM events e JOIN messages m ON m.id=e.subject AND m.id=json_extract(e.body,'$.message') + WHERE e.id=NEW.request AND e.business=NEW.business AND m.business=e.business AND e.kind='human.input' + AND e.actor_role IS NULL AND e.actor_run IS NULL AND json_extract(e.body,'$.kind')='instruction' AND m.from_role='human') + BEGIN SELECT RAISE(ABORT, 'request-origin-refused'); END; +CREATE TRIGGER request_events_revision BEFORE INSERT ON request_events + WHEN NEW.revision <> 1 + coalesce((SELECT max(revision) FROM request_events WHERE business=NEW.business AND request=NEW.request),0) + OR (NEW.revision=1) <> (NEW.kind='opened') + BEGIN SELECT RAISE(ABORT, 'request-revision-conflict'); END; +CREATE TRIGGER request_effects_binding BEFORE INSERT ON request_effects + WHEN NOT EXISTS (SELECT 1 FROM request_events WHERE business=NEW.business AND request=NEW.request) + OR EXISTS (SELECT 1 FROM request_effects WHERE business=NEW.business AND request=NEW.request AND op=NEW.op AND (digest<>NEW.digest OR kind<>NEW.kind)) + OR (NEW.stage<>'prepared' AND NOT EXISTS (SELECT 1 FROM request_effects WHERE business=NEW.business AND request=NEW.request AND op=NEW.op AND stage='prepared')) + OR EXISTS (SELECT 1 FROM request_effects WHERE business=NEW.business AND request=NEW.request AND op=NEW.op AND stage IN ('committed','refused')) + BEGIN SELECT RAISE(ABORT, 'request-effect-conflict'); END; +CREATE TRIGGER request_events_no_update BEFORE UPDATE ON request_events BEGIN SELECT RAISE(ABORT, 'request_events is append-only'); END; +CREATE TRIGGER request_events_no_delete BEFORE DELETE ON request_events BEGIN SELECT RAISE(ABORT, 'request_events is append-only'); END; +CREATE TRIGGER request_events_no_replace BEFORE INSERT ON request_events WHEN EXISTS (SELECT 1 FROM request_events WHERE seq=NEW.seq OR (business=NEW.business AND request=NEW.request AND revision=NEW.revision)) BEGIN SELECT RAISE(ABORT, 'request_events is append-only'); END; diff --git a/packages/bus/src/process.mjs b/packages/bus/src/process.mjs index 2ad612a1f..2cb679500 100644 --- a/packages/bus/src/process.mjs +++ b/packages/bus/src/process.mjs @@ -37,7 +37,7 @@ const LAUNCH_OPS = new Set([ 'sessionClearUncertain', 'sessionClearRetire', 'sessionClearStart', - 'request.admission', + 'request.admission', 'request.launches', 'request.launchResult', 'request.release', 'request.chief', 'request.manifest', ]); function launchOp(m) { if (m.op === 'identity') return runtime.identity(m.cap); @@ -82,6 +82,11 @@ function launchOp(m) { if (m.op === 'sessionClearStart') return runtime.sessionClearStart(m.record); // Lifecycle Task 2 (#797): may the launcher clone this project's code checkout. Scoped to the one // business and project named; a malformed ask answers launch-op-refused, never ready. + if (m.op === 'request.launches') return runtime.requestLaunches(m.business); + if (m.op === 'request.launchResult') return runtime.requestLaunchResult(m.record); + if (m.op === 'request.release') return runtime.requestRelease(m.cap, m.record); + if (m.op === 'request.manifest') return runtime.requestManifest(m.record); + if (m.op === 'request.chief') return runtime.requestChief(m.business); if (m.op === 'request.admission') return runtime.admission({ business: m.business, project: m.project }); if (m.op === 'credentialStatus') return runtime.credentialStatus(m.business, m.role); if (m.op === 'modelPreflight') return runtime.modelPreflight(m); diff --git a/packages/bus/src/request-continuation.mjs b/packages/bus/src/request-continuation.mjs new file mode 100644 index 000000000..39938d09b --- /dev/null +++ b/packages/bus/src/request-continuation.mjs @@ -0,0 +1,149 @@ +// Host-only fixed-stage continuations. Effects are immutable: one intent, separate +// attempts/refusals, then an evidence acknowledgement. Reading a message is never progress. +import { digest, refuse } from './request-store.mjs'; + +export function createContinuations({ journal, services }) { + const contextFor = (state, target) => { + const context = services.continuationContext(state, target); + const admission = services.requestAdmission?.(state) ?? null; + return { ...context, prerequisite: digest([context.prerequisite, admission]) }; + }; + const effects = state => journal.effects(state.business, state.request); + const unfinished = state => { + const rows = effects(state); + return rows.filter(e => e.kind === 'continuation' && e.stage === 'uncertain' && e.result?.intent && + !rows.some(r => r.op === e.op && r.stage === 'committed')); + }; + const receipt = (state, op, value, code = null) => { + const header = { version: 1, business: state.business, request: state.request, kind: 'continuation', op, digest: digest(value) }; + const old = journal.prepareEffect({ ...header, stage: 'prepared', result: null, code: null }); + if (old.stage !== 'prepared') return old; + return journal.finishEffect({ ...header, stage: 'committed', result: value, code }); + }; + function steps(state) { + if (state.phase === 'selection' && state.wait?.kind !== 'decision') return [[state.pm, 'intake_propose']]; + const binding = state.selection?.bindings.find(b => b.digest === state.plan_digest); + const staff = binding?.plan.staff; + if (!staff) return []; + if (state.phase === 'discovery') return [[state.pm, 'submit_prd']]; + if (state.phase === 'prd-review') return state.mode === 'autonomous' ? [[services.prdDecision?.(state.business, state.wait.subject)?.route_to ?? staff.technical_reviewer, 'resolve_decision']] : []; + if (state.phase === 'planning') return [[state.pm, 'task_create']]; + if (state.phase === 'implementation') return [[staff.coder, 'push_working'], [state.pm, 'advance_request']]; + if (state.phase === 'review') return [[staff.reviewer, 'review_verdict'], [state.pm, 'advance_request']]; + if (state.phase === 'delivery') return [[state.pm, 'task_close']]; + if (state.phase === 'complete') return [[staff.chief, 'send_message']]; + return []; + } + const evidence = state => ({ source_message: state.source_message, decision: state.wait?.kind === 'decision' ? state.wait.subject : null, + plan: state.plan_digest, spec_commit: state.spec_commit, candidate: state.candidate ? { prd_version: state.candidate.prd_version, records_commit: state.candidate.records_commit, prd_path: state.candidate.prd_path } : null, snapshot: state.snapshot, tasks: state.tasks, events: state.evidence ?? [] }); + const intentFor = (state, target, step) => ({ key: digest([state.request, state.revision, target, step]), + request: state.request, revision: state.revision, target, step, phase: state.phase, project: state.project, evidence: evidence(state) }); + function send(state, intent, context) { + const control = { request: state.request, revision: state.revision, op: `continue-${intent.key.slice(0, 32)}` }; + const next = { + intake_propose: { args: control, requires: ['source_project','project','name','tracker_project'] }, + submit_prd: { args: control, requires: ['prd_version','records_commit','prd_path','requirement_ids'] }, + resolve_decision: { args: { id: state.wait?.subject }, requires: ['choice','reason'] }, + task_create: { args: { request: state.request, project_key: state.project, requirement_snapshot: state.snapshot }, requires: ['title','requirement'] }, + push_working: { args: { task_ref: state.tasks[0], project: state.project }, requires: ['branch'] }, + review_verdict: { args: { task_ref: state.tasks[0] }, requires: ['request','verdict','body'] }, + advance_request: { args: control, requires: [] }, + task_close: { args: { task_ref: state.tasks[0] }, requires: ['verdict'] }, + send_message: { args: { in_reply_to: state.source_message }, requires: ['to','body'] }, + }[intent.step]; + const body = `Request continuation ${intent.key}\n${JSON.stringify({ ...intent, next: { tool: intent.step, ...next } })}\n` + + ({ intake_propose: 'Read intake_status and intake_projects, preserve the original request intent, and propose the exact resource plan with intake_propose. If the owner selected existing, select an accessible existing mapping. The owner confirms the plan; discovery needs no task or tracker effect.', + submit_prd: 'Read the committed spec and discovery records. Finish the product PRD and requirements with write_record, then submit_prd with their exact records commit, path, version and ids. Use request_staff for missing technical help.', + resolve_decision: 'Read the exact PRD candidate and requirements with intake_status(path), one selected path at a time. Resolve only the cited PRD decision through resolve_decision; ask for changes if evidence is incomplete.', + task_create: 'Read list_requirements for this project and snapshot. Check existing task/create receipts before creating or retrying. Create and assign the scoped tasks, request_staff for coder/reviewer sessions, then advance_request.', + push_working: 'Check the assigned tasks and existing receipts. Implement the approved requirements, push_working, and request_review for each latest commit.', + review_verdict: 'Fetch each current review request with fetch_review, review the artifacts and return review_verdict against that exact request and commit.', + advance_request: 'Check the existing task, push and review evidence, then call advance_request at the current intake_status revision. Never claim a success phase.', + task_close: 'Close each task with task_close citing the latest approve verdict. Verify the completed artifacts, then advance_request.', + send_message: 'Deliver the artifact and task/review evidence through the existing chief → assistant → user path, with send_message in_reply_to the original source_message so it returns to the originating session. Never coordinate the assistant directly with PM or reviewer.' }[intent.step]); + return services.continuationMessage(state, intent.target, body, context); + } + function attempt(state, row, context, initialMessage = null) { + const intent = row.result.intent, rows = effects(state); + const attempts = rows.filter(e => e.stage === 'committed' && e.result?.type === 'attempt' && e.result.key === intent.key); + const last = attempts.at(-1)?.result; + // Same role run means the same turn owns the work, even after it reads the notice. + // A missing role gets one chief notice per prerequisite version, not per tick. + const token = context.run ? `run:${context.run}` : `waiting:${context.prerequisite}`; + if (last?.token === token) return; + journal.transaction(() => { + let message = initialMessage; + if (!message && last?.message && (!context.run || services.continuationRouted?.(last.message, context.run))) { + message = last.message; + if (!context.run) services.continuationRetry?.(state, message, intent.target, context); + } + message ??= send(state, intent, context); + receipt(state, `attempt:${intent.key}:${attempts.length + 1}`, { type: 'attempt', key: intent.key, attempt: attempts.length + 1, + token, run: context.run, prerequisite: context.prerequisite, message }); + }); + } + function resume(state) { + if (!services.continuationContext || !services.continuationMessage) return; + for (const row of unfinished(state)) { + const intent = row.result.intent; + // Only a real lifecycle append can obsolete an intent. Launch/read events cannot. + const effect = services.continuationEffect?.(state, intent, row.result.message); + if (intent.phase !== state.phase || effect) { + journal.finishEffect({ ...row, stage: 'committed', result: { key: intent.key, evidence: { request: state.request, revision: state.revision, phase: state.phase, ...(effect ? { effect } : {}) } }, code: null }); + } + } + for (const [target, step] of steps(state)) { + const intent = intentFor(state, target, step), op = `continue:${intent.key}`; + let row = unfinished(state).find(e => e.op === op); + const context = contextFor(state, target); + if (!row) { + if (effects(state).some(e => e.op === op)) continue; + journal.transaction(() => { + const header = { version: 1, business: state.business, request: state.request, kind: 'continuation', op, digest: digest(intent) }; + journal.prepareEffect({ ...header, stage: 'prepared', result: null, code: null }); + const message = send(state, intent, context); + row = journal.finishEffect({ ...header, stage: 'uncertain', result: { intent, message }, code: null }); + attempt(state, row, context, message); + }); + } else attempt(state, row, context); + } + } + function launches(business) { + const out = []; + for (const r of journal.listRequests()) { + if (r.business !== business) continue; + const state = journal.readRequest(business, r.request); + for (const row of unfinished(state)) { + const intent = row.result.intent; + if (intent.revision !== state.revision) continue; + const context = contextFor(state, intent.target); + const lastChief = effects(state).findLast(e => e.stage === 'committed' && e.result?.type === 'launch' && e.result.key === intent.key && e.result.instance === context.chief); + out.push({ ...intent, ...context, chief_attempt: lastChief?.result?.prerequisite ?? null }); + } + } + return out; + } + function launchResult({ business, instance, code, run = null }) { + if (typeof business !== 'string' || typeof instance !== 'string' || (code !== null && !/^[a-z-]{1,128}$/.test(code)) || (run !== null && typeof run !== 'string')) refuse('invalid-request'); + for (const i of launches(business).filter(i => i.target === instance || i.chief === instance)) { + const state = journal.readRequest(business, i.request); + const value = { type: 'launch', key: i.key, instance, code, run, prerequisite: i.prerequisite }; + receipt(state, `launch:${digest(value)}`, value, code); + } + return { recorded: true }; + } + function manifestNeeded({ business, instance, project, run }) { + for (const i of launches(business).filter(i => i.target === instance && i.project === project && i.run === run)) { + const state = journal.readRequest(business, i.request), op = `manifest:${digest([i.key, run, project, i.chief_run, i.prerequisite])}`; + if (effects(state).some(e => e.op === op)) continue; + journal.transaction(() => { + const context = contextFor(state, i.chief); + const body = `Request continuation ${i.key}: request ${i.request}, revision ${i.revision}, phase ${i.phase}. Evidence ${JSON.stringify(i.evidence)}. Run ${run} lacks admitted project ${project} in its manifest. Next typed operation: launch({instance:${JSON.stringify(instance)}}). Let its current turn drain; use the existing launch/end path. Never interrupt unrelated work or hire a replacement.`; + const message = services.continuationMessage(state, i.chief, body, context); + receipt(state, op, { type: 'manifest', key: i.key, instance, project, run, message }); + }); + } + return { recorded: true }; + } + return Object.freeze({ resume, launches, launchResult, manifestNeeded }); +} diff --git a/packages/bus/src/request-lifecycle.mjs b/packages/bus/src/request-lifecycle.mjs index 5e1ce7eb0..2c0c2abe6 100644 --- a/packages/bus/src/request-lifecycle.mjs +++ b/packages/bus/src/request-lifecycle.mjs @@ -1,12 +1,14 @@ // Host controller for pre-task intake. Task 1 prepares and records selections only. // Later executors must consume these exact immutable bindings, never approvalChoice. +import { createContinuations } from './request-continuation.mjs'; import { lstatSync } from 'node:fs'; import { join, dirname } from 'node:path'; import { exact, object, id, key, text, digest, validatePlan, refuse } from './request-store.mjs'; export function createRequestLifecycle({ journal, services }) { if (!journal || !services || !['origin','authorize','raiseDecision','readDecision','project'].every(k => typeof services[k] === 'function')) refuse('request-services-unavailable'); - let closing = false; + const continuations = createContinuations({ journal, services }); + let closing = false, recovery = null, cursor = 0; const pending = new Map(); const serial = (business, request, fn) => { const k = business + '/' + request; @@ -41,10 +43,22 @@ export function createRequestLifecycle({ journal, services }) { }); } async function get(identity, args) { - exact(args, ['request']); if (!id(args.request)) refuse('invalid-request'); - const state = stateOf(identity.business, args.request); authorize(identity, state, 'get'); return state; + exact(args, ['request'], ['path']); if (!id(args.request) || (args.path !== undefined && !text(args.path, 256))) refuse('invalid-request'); + const state = stateOf(identity.business, args.request); authorize(identity, state, 'get'); + if (args.path === undefined) return state; + if (!state.candidate) refuse('prd-candidate-required'); + const selected = args.path === state.candidate.prd_path ? { blob: state.candidate.prd_blob } + : state.candidate.requirements.find(r => r.path === args.path); + if (!selected) refuse('request-document-refused'); + const material = await services.readCandidate(state.candidate, state.business); + const now = stateOf(state.business, state.request); authorize(identity, now, 'get'); + if (now.revision !== state.revision) refuse('request-revision-conflict'); + return { request: state.request, revision: state.revision, records_commit: state.candidate.records_commit, path: args.path, blob: selected.blob, + content: args.path === state.candidate.prd_path ? material.prd : material.requirements.find(r => r.id === selected.id) }; + } async function projects(identity, args) { + exact(args, ['request']); const state = await get(identity, args); const rows = await services.listProjects?.(structuredClone(state), identity) ?? []; authorize(identity, stateOf(state.business, state.request), 'get'); @@ -264,8 +278,7 @@ export function createRequestLifecycle({ journal, services }) { return stateOf(state.business, state.request); })); } - async function resume(business, request) { - return serial(business, request, async () => { + async function settle(business, request) { const state = stateOf(business, request); if (state.wait?.code === 'prd-approval') { try { await services.catalog.activate({ request, decision: state.wait.subject }); return stateOf(business, request); } @@ -300,9 +313,37 @@ export function createRequestLifecycle({ journal, services }) { return append(state, 'selection.confirmed', { decision: binding.decision, choice: binding.choice, digest: binding.digest }, null, tx.appendRequest); }); return runBootstrap(confirmed); + } + async function resume(business, request) { + return serial(business, request, async () => { + const state = await settle(business, request); + continuations.resume(state); + return stateOf(business, request); }); } - return Object.freeze({ open, get, projects, propose, submit, modeConsole, + async function advance(identity, args) { + exact(args, ['request','revision','op']); + if (!id(args.request) || !id(args.op) || !Number.isSafeInteger(args.revision) || args.revision < 1) refuse('invalid-request'); + return serial(identity.business, args.request, async () => { + let state = stateOf(identity.business, args.request); authorize(identity, state, 'advance'); + const proposal_digest = digest(args), old = journal.proposal(state.business, state.request, args.op); + if (old) { if (old.proposal_digest !== proposal_digest) refuse('request-operation-conflict'); return state; } + if (state.revision !== args.revision) refuse('request-revision-conflict'); + if (!state.snapshot || !services.progress) refuse('request-evidence-required'); + await services.catalog.resolve({ business: state.business, project: state.project, snapshot: state.snapshot }); + return journal.transaction(tx => { + state = stateOf(state.business, state.request); authorize(identity, state, 'advance'); + if (state.revision !== args.revision) refuse('request-revision-conflict'); + const progress = services.progress(state); + const result = append(state, 'advanced', { op: args.op, proposal_digest, from: state.phase, ...progress }, identity, tx.appendRequest); + services.record?.(identity, 'request.advance', result); + continuations.resume(result); + return result; + }); + }); + } + return Object.freeze({ open, get, projects, propose, submit, advance, modeConsole, + hasWork: () => journal.listRequests().length > 0, launches: continuations.launches, launchResult: continuations.launchResult, manifestNeeded: continuations.manifestNeeded, draftConsole: async (identity, args) => { exact(args, ['request']); if (!id(args.request)) refuse('invalid-request'); const state = stateOf(identity.business, args.request); authorize(identity, state, 'draft'); @@ -317,6 +358,21 @@ export function createRequestLifecycle({ journal, services }) { for (const row of journal.listRequests()) if (row.business === business && journal.readRequest(business, row.request).wait?.subject === decision) return resume(business, row.request); }, - close: async () => { closing = true; await Promise.allSettled([...pending.values()]); await services.catalog?.drain(); }, - recover: async () => { const states = []; for (const { business, request } of journal.listRequests()) states.push(await resume(business, request)); return states; } }); + close: async () => { closing = true; await recovery; await Promise.allSettled([...pending.values()]); await services.catalog?.drain(); }, + recover: ({ limit = Infinity } = {}) => { + if (closing) return Promise.resolve([]); + if (recovery) return recovery; + recovery = (async () => { + const states = [], rows = journal.listRequests(); + const count = Math.min(rows.length, limit), start = cursor % (rows.length || 1); + for (let n = 0; n < count; n++) { + if (closing) break; + const { business, request } = rows[(start + n) % rows.length]; + states.push(await resume(business, request)); + } + cursor = (start + count) % (rows.length || 1); + return states; + })().finally(() => { recovery = null; }); + return recovery; + } }); } diff --git a/packages/bus/src/request-store.mjs b/packages/bus/src/request-store.mjs index a37caeac5..e71129706 100644 --- a/packages/bus/src/request-store.mjs +++ b/packages/bus/src/request-store.mjs @@ -72,7 +72,7 @@ function initial(business, request, b) { owner: b.owner, pm: b.pm, phase: 'selection', project: null, chat_project: null, mode: null, mode_revision: 0, plan_digest: null, spec_commit: null, snapshot: null, tasks: [], delivery: null, question: null, candidate: null, selection: null, // The owner's chosen route survives every setup wait; a wait code never says which route was chosen. - route: null, wait: null }; + route: null, wait: null, launch_refusal: null }; } function reduce(state, e) { const code = 'invalid-request-event', b = e.body; @@ -148,6 +148,13 @@ function reduce(state, e) { s.mode = b.mode; s.mode_revision = b.mode_revision; s.candidate = null; s.snapshot = null; s.phase = 'discovery'; s.wait = { kind: 'setup', code: 'request-mode-changed', subject: s.request, resume_phase: s.phase, observed_revision: s.revision }; + } else if (e.kind === 'advanced') { + exact(b, ['op','proposal_digest','from','phase','tasks','evidence'], [], code); + const next = { planning: ['implementation'], implementation: ['review'], review: ['delivery','implementation'], delivery: ['complete'] }; + if (b.from !== s.phase || !next[s.phase]?.includes(b.phase) || !id(b.op) || !hash(b.proposal_digest) || + !Array.isArray(b.tasks) || !b.tasks.length || !b.tasks.every(x => /^vikunja:[1-9][0-9]*\/[1-9][0-9]*$/.test(x)) || + !Array.isArray(b.evidence) || !b.evidence.length || !b.evidence.every(id)) refuse(code); + s.tasks = [...b.tasks]; s.phase = b.phase; s.evidence = [...b.evidence]; s.wait = null; } else if (e.kind === 'cancelled') { exact(b, ['decision'], [], code); if (s.wait?.kind !== 'decision' || s.selection?.decision !== b.decision) refuse(code); @@ -170,7 +177,15 @@ export function createRequestStore(store) { if (!key(business) || !id(request)) refuse('invalid-request'); const events = store.all('SELECT * FROM request_events WHERE business=? AND request=? ORDER BY revision', business, request) .map(e => ({ ...e, body: JSON.parse(e.body) })); - return events.reduce((s, e) => { validateIntent(e, events[0].body); return reduce(s, e); }, null); + const state = events.reduce((s, e) => { validateIntent(e, events[0].body); return reduce(s, e); }, null); + if (!state) return state; + const launches = store.all("SELECT result FROM request_effects WHERE business=? AND request=? AND kind='continuation' AND stage='committed' AND json_extract(result,'$.type')='launch' ORDER BY seq", business, request).map(r => JSON.parse(r.result)); + const active = new Set(store.all("SELECT json_extract(result,'$.intent.key') key FROM request_effects WHERE business=? AND request=? AND kind='continuation' AND stage='uncertain' AND json_extract(result,'$.intent.revision')=?", business, request, state.revision).map(r => r.key)); + const latest = new Map(); + for (const launch of launches) if (active.has(launch.key)) latest.set(launch.instance, launch); + const last = launches.findLast(launch => launch.code && latest.get(launch.instance) === launch); + state.launch_refusal = last ? { instance: last.instance, code: last.code, key: last.key } : null; + return state; }; // appendInTransaction is only for composing the decision and journal atomically on the host. const appendInTransaction = e => { @@ -202,6 +217,9 @@ export function createRequestStore(store) { if (!d || d.action !== 'prd.approve' || d.project !== state.project || d.raised_by_role !== state.pm || d.raised_by_role !== e.actor.role || d.blocking !== 0 || d.task_ref !== null || d.requirement_ref !== null || d.route_to !== e.body.route.instance) refuse('invalid-request-event'); } + if (e.kind === 'advanced') { + if (e.actor.role !== state.pm || e.body.evidence.some(eid => !store.get('SELECT 1 FROM events WHERE business=? AND id=?', e.business, eid))) refuse('invalid-request-event'); + } if (e.kind === 'prd.activated') { const row = store.get('SELECT body FROM request_snapshots WHERE id=? AND business=? AND project=?', e.body.snapshot, e.business, state.project); if (!row || JSON.parse(row.body).approval.decision !== e.body.decision) refuse('invalid-request-event'); diff --git a/packages/bus/src/runtime.mjs b/packages/bus/src/runtime.mjs index 447501726..f9d5c8c32 100644 --- a/packages/bus/src/runtime.mjs +++ b/packages/bus/src/runtime.mjs @@ -282,7 +282,7 @@ export async function startBroker({ let modelProxyClient = null, proxyPage = null, proxyUsage = null, usageTimer = null; let modelKeys = null, modelTimer = null, modelSweep = null; const modelRuns = new Map(); - let store, credentials, server, adapter, processing, syncTimer = null, syncPass = null, accessTimer = null; + let store, credentials, server, adapter, processing, syncTimer = null, syncPass = null, accessTimer = null, requestTimer = null; try { // Lifecycle Task 2 (#797): a product whose mapping is published but not yet activated is not part of // this process. The request journal says which are activated; every other managed entry waits for the @@ -575,7 +575,7 @@ export async function startBroker({ const catalog = createRequirementCatalog({ journal: { ...requestJournal, readCandidate, authority: services.prdAuthority, approval: services.prdApproval }, readSpec: ({ business, project }) => projectModule.readSpec({ business: services.prdBusiness(business), project, dataRoot }), builtinBinding: null }); - return (requestLifecycle = createRequestLifecycle({ journal: requestJournal, services: { catalog, readCandidate, ...(writer ? hostPreparation({ dataRoot, organization: writer, trackers }) : {}), ...requestServices, ...services, + return (requestLifecycle = createRequestLifecycle({ journal: requestJournal, services: { catalog, readCandidate, requestAdmission: state => admission({ business: state.business, project: state.project }), ...(writer ? hostPreparation({ dataRoot, organization: writer, trackers }) : {}), ...requestServices, ...services, // The new-project executor needs the organization writer; without one a confirmed selection stays at its setup wait. ...(writer ? { bootstrap: ({ binding, state }) => processing.work(() => executeBootstrap({ binding, owner: state.owner, journal: requestJournal, services: { dataRoot, organization: writer, activate: activateManagedProject, barrier: requestServices.barrier, @@ -972,11 +972,22 @@ export async function startBroker({ usageTimer.unref(); usageTick(); // Optional telemetry never blocks the native ready answer. await requestLifecycle.recover(); + requestTimer = setInterval(() => { + const state = processing.read(); + if (state.halted || state.invalid) return; + requestLifecycle.recover({ limit: 64 }).catch(e => process.stderr.write(`request: ${JSON.stringify({ event: 'reconcile-refused', code: e.code ?? 'request-failed' })}\n`)); + }, 30000); + requestTimer.unref(); return { broker, requestLifecycle, requirementCatalog: requestLifecycle.catalog, admission, + requestLaunches: business => requestLifecycle.launches(business), + requestLaunchResult: record => requestLifecycle.launchResult(record), + requestManifest: record => requestLifecycle.manifestNeeded(record), + requestRelease: (cap, record) => refusing(cap, () => broker.requestRelease(cap, record)), + requestChief: business => broker.requestChief(business), credentials, // P2 consumes this inside the bus only. No process IPC/verb returns it. get modelProxy() { return modelProxyClient; }, @@ -1038,6 +1049,7 @@ export async function startBroker({ launches: bound, readers: readCaps, async close() { + clearInterval(requestTimer); clearInterval(usageTimer); await proxyUsage.close(); clearInterval(modelTimer); @@ -1053,6 +1065,7 @@ export async function startBroker({ }, }; } catch (e) { + clearInterval(requestTimer); clearInterval(usageTimer); await proxyUsage?.close(); clearInterval(modelTimer); diff --git a/packages/bus/src/store.mjs b/packages/bus/src/store.mjs index 5777124fd..9266aafad 100644 --- a/packages/bus/src/store.mjs +++ b/packages/bus/src/store.mjs @@ -13,8 +13,8 @@ import { join, isAbsolute } from 'node:path'; import { createHash } from 'node:crypto'; const ddl = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8'); -export const SCHEMA_SHA256 = '2a6a161b3d671418e690942f02203b4a014e939b46ae8f651da11f091dac5455'; -export const SCHEMA_VERSION = '3q'; +export const SCHEMA_SHA256 = '24211532d78c1c5fbac655197fed2fb30986f8d7f88c7ef77cdec2895b22400d'; +export const SCHEMA_VERSION = '3r'; // Ordered, named offline steps. New schema changes append a step and keep prior DDL pinned. const legacyDDL = readFileSync(new URL('./migrations/3b.sql', import.meta.url), 'utf8'); const LEGACY_SHA256 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65'; @@ -96,6 +96,9 @@ const migrations = [{ }, { from: '3p', to: '3q', sql: readFileSync(new URL('./migrations/3p-3q.sql', import.meta.url), 'utf8'), +}, { + from: '3q', to: '3r', + sql: readFileSync(new URL('./migrations/3q-3r.sql', import.meta.url), 'utf8'), }]; const timedTables = [ 'events', diff --git a/packages/bus/tests/account-migration.test.mjs b/packages/bus/tests/account-migration.test.mjs index 7f2e594ba..8e4ceeea9 100644 --- a/packages/bus/tests/account-migration.test.mjs +++ b/packages/bus/tests/account-migration.test.mjs @@ -91,6 +91,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) { copyFileSync(here('../src/migrations/3n-3o.sql'), join(moduleRoot, 'src/migrations/3n-3o.sql')); copyFileSync(here('../src/migrations/3o-3p.sql'), join(moduleRoot, 'src/migrations/3o-3p.sql')); copyFileSync(here('../src/migrations/3p-3q.sql'), join(moduleRoot, 'src/migrations/3p-3q.sql')); + copyFileSync(here('../src/migrations/3q-3r.sql'), join(moduleRoot, 'src/migrations/3q-3r.sql')); return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store; } @@ -98,7 +99,7 @@ test('REQ-LIMIT-4 3h copy migrates to 3i: every row keeps its seq, the sequence const { root, path } = source3h(t), before = snapshot(path), was = master(path); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); // Every table's rows as they were, usage_records with its seq values and sqlite_sequence with 50. assert.deepEqual(migrated3l(snapshot(path, before.columns).rows), before.rows); assert.deepEqual(before.rows.usage_records.map(r => r.seq), [1, 2, 3]); @@ -148,7 +149,7 @@ test('REQ-LIMIT-4 a 3h→3i step that fails partway leaves the store at 3h with assert.deepEqual(snapshot(path, before.columns).rows, before.rows); // Still a whole 3h store: the real step migrates it. new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); assert.deepEqual(migrated3l(snapshot(path, before.columns).rows), before.rows); }); @@ -160,14 +161,14 @@ test('REQ-LIMIT-4 the 3h engine refuses a migrated store, migrating or not, and schema: ddl3h, store: (source) => source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3h).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3h';") + .replace("export const SCHEMA_VERSION = '3r';", "export const SCHEMA_VERSION = '3h';") .replace(/\}, \{\n {2}\/\/ Row 126 \(#1620\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'), }); const pinned = readFileSync(join(root, 'engine-3h/src/store.mjs'), 'utf8'); assert.ok(pinned.includes("export const SCHEMA_VERSION = '3h';") && !pinned.includes('3h-3i'), 'the 3h engine has no 3h→3i step'); assert.throws(() => new Store3h(root), /schema-mismatch/); assert.throws(() => new Store3h(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); assert.deepEqual(master(path), was); assert.deepEqual(snapshot(path, after.columns).rows, after.rows); }); diff --git a/packages/bus/tests/chief-of-staff.test.mjs b/packages/bus/tests/chief-of-staff.test.mjs index 7c8566b5e..aa5f6261e 100644 --- a/packages/bus/tests/chief-of-staff.test.mjs +++ b/packages/bus/tests/chief-of-staff.test.mjs @@ -614,7 +614,7 @@ async function database3c(t) { let source = readFileSync(here('../src/store.mjs'), 'utf8'); const pinned = source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3c).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3c';") + .replace("export const SCHEMA_VERSION = '3r';", "export const SCHEMA_VERSION = '3c';") .replace(/\}, \{\n {2}from: '3c',[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); assert.notEqual(pinned, source); assert.ok(!pinned.includes('3c-3d'), 'the 3c engine has no 3c→3d step'); @@ -638,7 +638,7 @@ test('REQ-ROLE-2, REQ-COS-1: the 3c→3d migration lets the chief of staff end a assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); t.after(() => store.close()); - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3r'); assert.equal(store.get("SELECT body FROM events WHERE id='preserved'").body, '{"fixture":true}'); assert.equal(store.get("SELECT holder_run FROM role_claims WHERE role='coder'").holder_run, 'coder-run'); const b = new Broker({ store, businesses: demo(), organization: null }); diff --git a/packages/bus/tests/comment-migration.test.mjs b/packages/bus/tests/comment-migration.test.mjs index f08be9c56..1451f882e 100644 --- a/packages/bus/tests/comment-migration.test.mjs +++ b/packages/bus/tests/comment-migration.test.mjs @@ -33,7 +33,7 @@ test('explicit 3b copy migration preserves evidence and opens as 3g; normal star const original = readFileSync(source); assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3r'); assert.equal(store.get("SELECT body FROM events WHERE id='preserved'").body, '{"fixture":true}'); assert.equal(store.get('SELECT count(*) n FROM task_comment_cursors').n, 0); const epoch = store.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'").value; @@ -75,15 +75,15 @@ test('named migration registry chains 3b through 3g to a fixture-only future ver const sha = createHash('sha256').update(futureDDL).digest('hex'); let source = readFileSync(new URL('../src/store.mjs', import.meta.url), 'utf8'); source = source.replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${sha}';`) - .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3r';") - .replace('}];\nconst timedTables', `}, { from: '3q', to: '3r', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`); + .replace("export const SCHEMA_VERSION = '3r';", "export const SCHEMA_VERSION = '3s';") + .replace('}];\nconst timedTables', `}, { from: '3r', to: '3s', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`); writeFileSync(join(moduleRoot, 'src/store.mjs'), source); writeFileSync(join(moduleRoot, 'schema.sql'), futureDDL); - for (const name of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql', '3g-3h.sql', '3h-3i.sql', '3i-3j.sql', '3j-3k.sql', '3k-3l.sql', '3l-3m.sql', '3m-3n.sql', '3n-3o.sql', '3o-3p.sql', '3p-3q.sql']) + for (const name of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql', '3g-3h.sql', '3h-3i.sql', '3i-3j.sql', '3j-3k.sql', '3k-3l.sql', '3l-3m.sql', '3m-3n.sql', '3n-3o.sql', '3o-3p.sql', '3p-3q.sql', '3q-3r.sql']) copyFileSync(new URL('../src/migrations/' + name, import.meta.url), join(moduleRoot, 'src/migrations', name)); const { Store: FutureStore } = await import(pathToFileURL(join(moduleRoot, 'src/store.mjs'))); const future = new FutureStore(root, { migrateSchema: true }); - assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3r'); + assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3s'); assert.equal(future.get("SELECT name FROM sqlite_master WHERE name='future_fixture'").name, 'future_fixture'); assert.equal(future.get("SELECT count(*) n FROM task_comment_cursors").n, 0); assert.equal(future.get("SELECT id FROM events").id, 'preserved'); @@ -109,7 +109,7 @@ test('offline command migrates only the supplied fixture root and reports usage const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname; const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' }); assert.equal(migrated.status, 0, migrated.stderr); - assert.match(migrated.stdout, /bus schema 3q ready/); + assert.match(migrated.stdout, /bus schema 3r ready/); new Store(root).close(); const usage = spawnSync(process.execPath, [command], { encoding: 'utf8' }); assert.equal(usage.status, 2); diff --git a/packages/bus/tests/gitea-migration.test.mjs b/packages/bus/tests/gitea-migration.test.mjs index 17a280340..c7b7a9a8c 100644 --- a/packages/bus/tests/gitea-migration.test.mjs +++ b/packages/bus/tests/gitea-migration.test.mjs @@ -81,6 +81,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) { copyFileSync(here('../src/migrations/3n-3o.sql'), join(moduleRoot, 'src/migrations/3n-3o.sql')); copyFileSync(here('../src/migrations/3o-3p.sql'), join(moduleRoot, 'src/migrations/3o-3p.sql')); copyFileSync(here('../src/migrations/3p-3q.sql'), join(moduleRoot, 'src/migrations/3p-3q.sql')); + copyFileSync(here('../src/migrations/3q-3r.sql'), join(moduleRoot, 'src/migrations/3q-3r.sql')); return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store; } function fresh(t) { @@ -102,7 +103,7 @@ test('3j copy migrates to 3k: every row kept, and sqlite_master gains only the t const { root, path } = source3j(t), before = rows(path), was = master(path); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); assert.deepEqual(migrated3l(rows(path)), before); const now = master(path); assert.deepEqual(now.filter(r => !NEW.includes(r.name) && !later3l(r)), was.filter(r => !later3l(r))); @@ -128,7 +129,7 @@ test('a 3j→3k step that fails partway leaves the store at 3j with every row an assert.deepEqual(master(path), was); assert.deepEqual(rows(path), before); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); }); test('the 3j engine refuses a migrated store, migrating or not, and leaves it as it was', async t => { @@ -139,14 +140,14 @@ test('the 3j engine refuses a migrated store, migrating or not, and leaves it as schema: ddl3j, store: (source) => source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3j).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3j';") + .replace("export const SCHEMA_VERSION = '3r';", "export const SCHEMA_VERSION = '3j';") .replace(/\}, \{\n {2}\/\/ Row 236 \(#1742[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'), }); const pinned = readFileSync(join(root, 'engine-3j/src/store.mjs'), 'utf8'); assert.ok(pinned.includes("export const SCHEMA_VERSION = '3j';") && !pinned.includes('3j-3k'), 'the 3j engine has no 3j→3k step'); assert.throws(() => new Store3j(root), /schema-mismatch/); assert.throws(() => new Store3j(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); assert.deepEqual(master(path), was); assert.deepEqual(rows(path), after); }); diff --git a/packages/bus/tests/grant-migration.test.mjs b/packages/bus/tests/grant-migration.test.mjs index 5552d16bd..8fe1d66d0 100644 --- a/packages/bus/tests/grant-migration.test.mjs +++ b/packages/bus/tests/grant-migration.test.mjs @@ -84,6 +84,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) { copyFileSync(here('../src/migrations/3n-3o.sql'), join(moduleRoot, 'src/migrations/3n-3o.sql')); copyFileSync(here('../src/migrations/3o-3p.sql'), join(moduleRoot, 'src/migrations/3o-3p.sql')); copyFileSync(here('../src/migrations/3p-3q.sql'), join(moduleRoot, 'src/migrations/3p-3q.sql')); + copyFileSync(here('../src/migrations/3q-3r.sql'), join(moduleRoot, 'src/migrations/3q-3r.sql')); return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store; } function fresh(t) { @@ -107,7 +108,7 @@ test('3i copy migrates to 3j: every row kept, and sqlite_master gains only the t const { root, path } = source3i(t), before = rows(path), was = master(path); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); assert.deepEqual(migrated3l(rows(path)), before); const now = master(path).filter(r => !LATER.includes(r.name)); assert.deepEqual(now.filter(r => !NEW.includes(r.name) && !later3l(r)), was.filter(r => !later3l(r))); @@ -132,7 +133,7 @@ test('a 3i→3j step that fails partway leaves the store at 3i with every row an assert.deepEqual(master(path), was); assert.deepEqual(rows(path), before); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); }); test('the 3i engine refuses a migrated store, migrating or not, and leaves it as it was', async t => { @@ -143,14 +144,14 @@ test('the 3i engine refuses a migrated store, migrating or not, and leaves it as schema: ddl3i, store: (source) => source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3i).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3i';") + .replace("export const SCHEMA_VERSION = '3r';", "export const SCHEMA_VERSION = '3i';") .replace(/\}, \{\n {2}\/\/ Row 234 \(#1740\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'), }); const pinned = readFileSync(join(root, 'engine-3i/src/store.mjs'), 'utf8'); assert.ok(pinned.includes("export const SCHEMA_VERSION = '3i';") && !pinned.includes('3i-3j'), 'the 3i engine has no 3i→3j step'); assert.throws(() => new Store3i(root), /schema-mismatch/); assert.throws(() => new Store3i(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); assert.deepEqual(master(path), was); assert.deepEqual(rows(path), after); }); diff --git a/packages/bus/tests/model-hire-ownership-fixture.mjs b/packages/bus/tests/model-hire-ownership-fixture.mjs index a8da4a87c..c675e8a16 100644 --- a/packages/bus/tests/model-hire-ownership-fixture.mjs +++ b/packages/bus/tests/model-hire-ownership-fixture.mjs @@ -77,6 +77,11 @@ export async function ownershipFixture(t,{launches=true,listed=true,keep=false,r case 'lifecycleDelete':return bus.lifecycleDelete(m.record); case 'lifecycleDone':return bus.lifecycleDone(m.record); case 'sessionClears':return bus.sessionClears(m.business); + case 'request.launches':return bus.requestLaunches(m.business); + case 'request.launchResult':return bus.requestLaunchResult(m.record); + case 'request.chief':return bus.requestChief(m.business); + case 'request.manifest':return bus.requestManifest(m.record); + case 'request.admission':return bus.admission(m); default:throw Error('fixture unrequested trusted operation'); } }; diff --git a/packages/bus/tests/name-migration.test.mjs b/packages/bus/tests/name-migration.test.mjs index 6855952e2..579a70fbd 100644 --- a/packages/bus/tests/name-migration.test.mjs +++ b/packages/bus/tests/name-migration.test.mjs @@ -65,7 +65,7 @@ test('3f copy migrates to 3g preserving every old column/row, with an empty inst assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); try { - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3r'); assert.deepEqual({ ...store.get('SELECT to_name FROM messages') }, { to_name: null }); assert.equal(store.get('SELECT count(*) AS n FROM instance_names').n, 0); assert.deepEqual({ ...store.get('SELECT spec_path,spec_commit FROM decisions') }, { spec_path: 'docs/spec.md', spec_commit: 'c'.repeat(40) }); diff --git a/packages/bus/tests/prd-activation.test.mjs b/packages/bus/tests/prd-activation.test.mjs index 52ad61c80..08ad924ef 100644 --- a/packages/bus/tests/prd-activation.test.mjs +++ b/packages/bus/tests/prd-activation.test.mjs @@ -173,7 +173,7 @@ test('3p to 3q migration preserves existing request rows and the append-only gua const { Store, schemaDigest } = await import('../src/store.mjs'); const root = tmp(t, 'prd-migration-'); mkdirSync(join(root, 'bus'), { mode: 0o700 }); const path = join(root, 'bus/bus.sqlite'), ddl = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8') - .replace(",'prd.submitted','prd.activated','prd.invalidated','mode.changed'", ''); + .replace(",'prd.submitted','prd.activated','prd.invalidated','mode.changed','advanced'", ''); const db = new DatabaseSync(path); db.exec(ddl); const at = '2026-11-03T00:00:00.000Z'; db.prepare('INSERT INTO meta VALUES(?,?)').run('schema_version', '3p'); @@ -188,7 +188,7 @@ test('3p to 3q migration preserves existing request rows and the append-only gua assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); t.after(() => store.close()); assert.deepEqual(store.all('SELECT * FROM request_events'), before); - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3r'); store.run("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','r',2,'prd.submitted',?,'{}')", at); assert.throws(() => store.run('DELETE FROM request_events'), /append-only/); assert.throws(() => store.run("UPDATE request_events SET body='{}'"), /append-only/); diff --git a/packages/bus/tests/request-continuation.test.mjs b/packages/bus/tests/request-continuation.test.mjs new file mode 100644 index 000000000..00a341354 --- /dev/null +++ b/packages/bus/tests/request-continuation.test.mjs @@ -0,0 +1,249 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { requestFixture } from './request-fixture.mjs'; + +const intents = f => f.journal.effects('demo', f.request).filter(e => e.kind === 'continuation' && e.stage === 'uncertain' && e.result?.intent); +const messages = (f, key) => f.store.all("SELECT * FROM messages WHERE from_role='_bus' AND body LIKE ?", `%${key}%`); +const advance = async (f, op = 'next') => { + const s = f.journal.readRequest('demo', f.request); + return f.call('pm', 'request.advance', { request: f.request, revision: s.revision, op }); +}; + +test('resume atomically journals one intent and message; read is not an acknowledgement; restart keeps the key', async t => { + const f = await requestFixture(t); + await f.bus.requestLifecycle.recover(); + const [intent] = intents(f); + assert.ok(intent, 'discovery has a durable continuation'); + assert.equal(intent.result.intent.target, 'pm'); + assert.equal(intent.result.intent.step, 'submit_prd'); + const key = intent.result.intent.key; + assert.match(messages(f, key)[0].body, /source_message|evidence/); + for (let i = 0; i < 3; i++) await f.bus.requestLifecycle.recover(); + assert.equal(messages(f, key).length, 1); + const received = f.call('pm', 'message.receive'); + for (const m of received) f.call('pm', 'message.read', { id: m.id }); + await f.bus.requestLifecycle.recover(); + assert.ok(!f.journal.effects('demo', f.request).some(e => e.op === intent.op && e.stage === 'committed')); + await f.bus.close(); + const next = await f.boot(); + assert.equal(f.journal.effects('demo', f.request).filter(e => e.op === intent.op && e.stage === 'prepared').length, 1); + await next.requestLifecycle.recover(); + assert.equal(messages(f, key).length, 1); +}); + +test('a dead role after delivery receives the same unfinished key in its successor, once', async t => { + const f = await requestFixture(t); + await f.bus.requestLifecycle.recover(); + const intent = intents(f)[0], key = intent.result.intent.key; + f.call('pm', 'message.receive'); + f.bus.broker.endLaunch({ business: 'demo', run: 'pm-run', reason: 'exit', exitCode: 1 }); + f.caps.pm = f.bus.broker.bindLaunch({ business: 'demo', role: 'pm', run: 'pm-successor', harness: 'pi' }); + f.call('pm', 'role.claim'); + await f.bus.requestLifecycle.recover(); + assert.ok(f.call('pm', 'message.receive').some(m => m.body.includes(key))); + await f.bus.requestLifecycle.recover(); + assert.equal(messages(f, key).filter(m => m.to_role === 'pm').length, 2); + assert.equal(intents(f).filter(e => e.op === intent.op).length, 1); +}); + +test('wrong-role and claimed state advances fail; planning needs actual task evidence', async t => { + const f = await requestFixture(t), sub = await f.submit(); f.approve(sub); await f.activate(sub); + const s = f.journal.readRequest('demo', f.request), args = { request: f.request, revision: s.revision, op: 'next' }; + for (const role of ['cto','coder','reviewer','chief','human']) + await assert.rejects(Promise.resolve().then(() => f.call(role, 'request.advance', args)), /request-access-refused|agent-required/); + await assert.rejects(f.call('pm', 'request.advance', { ...args, phase: 'complete' }), /invalid-request/); + await assert.rejects(advance(f), /request-evidence-required/); + assert.equal(f.journal.readRequest('demo', f.request).phase, 'planning'); +}); + +test('validated PRD transition acknowledges discovery and schedules planning without flooding', async t => { + const f = await requestFixture(t); await f.bus.requestLifecycle.recover(); + const discovery = intents(f)[0]; + const sub = await f.submit(); f.approve(sub); await f.activate(sub); + await f.bus.requestLifecycle.recover(); + assert.ok(f.journal.effects('demo', f.request).some(e => e.op === discovery.op && e.stage === 'committed' && e.result.evidence)); + assert.ok(intents(f).some(e => e.result.intent.step === 'task_create')); +}); + +test('exact launch refusal is durable and returned by intake status, with bounded retries', async t => { + const f = await requestFixture(t); await f.bus.requestLifecycle.recover(); + await f.bus.requestLifecycle.launchResult({ business: 'demo', instance: 'pm', code: 'capacity-full', run: null }); + const before = f.store.get('SELECT count(*) n FROM messages').n; + for (let i = 0; i < 5; i++) await f.bus.requestLifecycle.recover(); + const state = await f.call('pm', 'request.get', { request: f.request }); + assert.equal(state.launch_refusal.code, 'capacity-full'); + assert.equal(state.launch_refusal.instance, 'pm'); + assert.equal(f.store.get('SELECT count(*) n FROM messages').n, before); +}); + +async function planned(t, threaded = false) { + const f = await requestFixture(t, 'autonomous', null, threaded), sub = await f.submit(); f.approve(sub); await f.activate(sub); + const { createDigest, createMarker } = await import('../src/broker.mjs'); + const s = f.journal.readRequest('demo', f.request), task_ref = 'vikunja:2/41'; + const prepared = { op: 'timer-task', request: f.request, requirement: 'REQ-POMO-1', project_key: 'timer', requirement_snapshot: s.snapshot, + project: 2, creator: 3, decision: null, view: { ids: { todo: 1, review: 2, done: 3 } }, labels: [], relations: [] }; + prepared.body = { title: 'Timer', description: createMarker(prepared) }; + prepared.body_digest = createDigest(prepared.body); + const { receipt } = f.bus.broker.prepareTaskCreate(f.caps.pm, prepared); + let fields = { ...prepared.body, project_id: 2, bucket: 1, done: false, assignees: [], due_date: null, priority: 0 }; + const snapshot = () => ({ task_ref, fields, digest: createDigest(fields), updated: new Date().toISOString() }); + f.bus.broker.recordRecoveredCreate({ receipt, task_ref, snapshot: { ...snapshot(), creator: 3 } }); + f.bus.broker.recordTask({ cap: f.caps.pm, business: 'demo', snapshots: [snapshot()], events: [] }); + await assert.rejects(advance(f), /request-evidence-required/); + fields = { ...fields, assignees: [5] }; + f.bus.broker.recordTask({ cap: f.caps.pm, business: 'demo', snapshots: [snapshot()], events: [{ kind: 'task.assigned', subject: task_ref, body: { role: 'coder', previous: [] } }] }); + const push = commit => f.bus.broker.recordPush(f.caps.coder, { task_ref, project: 'timer', branch: 'timer', ref: 'work/vikunja-2-41', commit, previous: null }); + const review = commit => f.call('coder', 'review.request', { task_ref, commit, to: 'reviewer' }); + return { ...f, task_ref, push, review, closeTask: verdict => { + fields = { ...fields, done: true, bucket: 3 }; + f.bus.broker.recordTask({ cap: f.caps.pm, business: 'demo', snapshots: [snapshot()], events: [{ kind: 'task.closed', subject: task_ref, body: { verdict } }] }); + } }; +} + +test('fixed advance derives implementation, review, delivery and completion from actual scoped receipts', async t => { + const f = await planned(t); + const initial = f.journal.readRequest('demo', f.request); + const args = { request: f.request, revision: initial.revision, op: 'start' }; + const implementation = await f.call('pm', 'request.advance', args); + assert.equal(implementation.phase, 'implementation'); + assert.deepEqual(await f.call('pm', 'request.advance', args), implementation); + await assert.rejects(f.call('pm', 'request.advance', { ...args, revision: initial.revision + 1 }), /request-operation-conflict/); + await assert.rejects(advance(f, 'before-push'), /request-evidence-required/); + const commit = 'b'.repeat(40); f.push(commit); + await assert.rejects(advance(f, 'before-review-request'), /request-evidence-required/); + const requested = f.review(commit); + assert.equal((await advance(f, 'review')).phase, 'review'); + await assert.rejects(advance(f, 'before-verdict'), /request-evidence-required/); + const verdict = f.call('reviewer', 'review.verdict', { task_ref: f.task_ref, request: requested.request, verdict: 'approve', body: 'Artifact verified' }); + assert.equal((await advance(f, 'delivery')).phase, 'delivery'); + await assert.rejects(advance(f, 'before-close'), /request-evidence-required/); + f.closeTask(verdict.verdict); + assert.equal((await advance(f, 'complete')).phase, 'complete'); +}); + +test('a newer push cannot borrow an older approved verdict; changes resume implementation', async t => { + const f = await planned(t); await advance(f, 'start'); + const first = 'b'.repeat(40), second = 'c'.repeat(40); + f.push(first); const r = f.review(first); await advance(f, 'review'); + f.call('reviewer', 'review.verdict', { task_ref: f.task_ref, request: r.request, verdict: 'approve', body: 'First approved' }); + f.push(second); + await assert.rejects(advance(f, 'stale'), /request-evidence-required/); + const changed = f.review(second); + f.call('reviewer', 'review.verdict', { task_ref: f.task_ref, request: changed.request, verdict: 'changes', body: 'Fix artifact' }); + assert.equal((await advance(f, 'changes')).phase, 'implementation'); +}); + +test('a failed message insert rolls back the continuation intent and attempts as one transaction', async t => { + const f = await requestFixture(t), { DatabaseSync } = await import('node:sqlite'), { join } = await import('node:path'); + const db = new DatabaseSync(join(f.f.dataRoot, 'bus/bus.sqlite')); t.after(() => db.close()); + // Move to a new validated stage before injecting the message failure. + const sub = await f.submit(); f.approve(sub); await f.activate(sub); + db.exec("CREATE TRIGGER fail_continuation BEFORE INSERT ON messages WHEN NEW.from_role='_bus' AND NEW.body LIKE 'Request continuation %' BEGIN SELECT RAISE(ABORT,'fixture-send-failed'); END"); + const count = f.store.get("SELECT count(*) n FROM request_effects WHERE stage='prepared'").n; + await assert.rejects(f.bus.requestLifecycle.recover(), /fixture-send-failed/); + assert.equal(f.store.get("SELECT count(*) n FROM request_effects WHERE stage='prepared'").n, count); + db.exec('DROP TRIGGER fail_continuation'); + await f.bus.requestLifecycle.recover(); + assert.equal(intents(f).filter(e => e.result.intent.step === 'task_create').length, 1); +}); + +test('a waiting message is routed to the first holder without duplicate delivery at reconciliation', async t => { + const f = await requestFixture(t); + f.bus.broker.endLaunch({ business: 'demo', run: 'pm-run', reason: 'exit', exitCode: 0 }); + await f.bus.requestLifecycle.recover(); + const intent = intents(f)[0], key = intent.result.intent.key; + const before = messages(f, key).filter(m => m.to_role === 'pm').length; + f.caps.pm = f.bus.broker.bindLaunch({ business: 'demo', role: 'pm', run: 'first-live-pm', harness: 'pi' }); f.call('pm', 'role.claim'); + await f.bus.requestLifecycle.recover(); + assert.equal(messages(f, key).filter(m => m.to_role === 'pm').length, before); + assert.equal(f.call('pm', 'message.receive').filter(m => m.body.includes(key)).length, 1); +}); + +test('3q to 3r migration preserves journal and sequence and admits only the new checked event', async t => { + const { DatabaseSync } = await import('node:sqlite'); + const { Store, schemaDigest } = await import('../src/store.mjs'); + const { readFileSync, mkdirSync, chmodSync } = await import('node:fs'); + const { join } = await import('node:path'); + const { tmp } = await import('../../cli/tests/helpers.mjs'); + const root = tmp(t, 'continuation-migration-'); mkdirSync(join(root, 'bus'), { mode: 0o700 }); + const path = join(root, 'bus/bus.sqlite'); + const db = new DatabaseSync(path); + db.exec(readFileSync(new URL('../schema.sql', import.meta.url), 'utf8').replace(",'advanced'", '')); + db.prepare('INSERT INTO meta VALUES(?,?)').run('schema_version', '3q'); + db.prepare('INSERT INTO meta VALUES(?,?)').run('schema_digest', schemaDigest(db)); + db.prepare('INSERT INTO meta VALUES(?,?)').run('comment_cursor_epoch', '2026-11-04T00:00:00.000Z'); + const at = '2026-11-04T00:00:00.000Z'; + db.prepare("INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES('m',?,'demo','human','human','pm','INFO','Timer')").run(at); + db.prepare("INSERT INTO events(id,at,business,kind,subject,body) VALUES('r',?,'demo','human.input','m',?)").run(at, JSON.stringify({ kind: 'instruction', message: 'm' })); + db.prepare("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','r',1,'opened',?,'{}')").run(at); + const before = db.prepare('SELECT * FROM request_events').all(); db.close(); chmodSync(path, 0o600); + assert.throws(() => new Store(root), /schema-mismatch/); + const store = new Store(root, { migrateSchema: true }); t.after(() => store.close()); + assert.deepEqual(store.all('SELECT * FROM request_events'), before); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3r'); + store.run("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','r',2,'advanced',?,'{}')", at); + assert.equal(store.get("SELECT max(seq) n FROM request_events").n, before[0].seq + 1); + assert.throws(() => store.run('DELETE FROM request_events'), /append-only/); +}); + +test('the bounded runtime reconciliation catches a missed notification after 30 seconds', async t => { + t.mock.timers.enable({ apis: ['setInterval'] }); + const f = await requestFixture(t); await f.bus.requestLifecycle.recover(); + const { DatabaseSync } = await import('node:sqlite'), { join } = await import('node:path'); + const db = new DatabaseSync(join(f.f.dataRoot, 'bus/bus.sqlite')); t.after(() => db.close()); + const s = f.journal.readRequest('demo', f.request), count = intents(f).length; + // Trusted journal evidence with deliberately no broker notification, as after a missed IPC wakeup. + db.prepare("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo',?,?,'setup.wait',?,?)") + .run(f.request, s.revision + 1, new Date().toISOString(), JSON.stringify({ code: 'fixture-prerequisite-cleared', subject: f.request })); + t.mock.timers.tick(29999); + await new Promise(r => setImmediate(r)); + assert.equal(intents(f).length, count); + t.mock.timers.tick(1); + await new Promise(r => setImmediate(r)); await new Promise(r => setImmediate(r)); + assert.equal(intents(f).length, count + 1); + const old = intents(f).find(i => i.result.intent.revision === s.revision); + assert.ok(!f.journal.effects('demo', f.request).some(e => e.op === old.op && e.stage === 'committed'), 'a setup wait is not progress acknowledgement'); +}); + +test('eligible CTO reads one exact draft document before tasks; arbitrary paths and writers stay refused', async t => { + const f = await requestFixture(t), sub = await f.submit(); + const args = { request: f.request, path: sub.candidate.prd_path }; + const draft = await f.call('cto', 'request.get', args); + assert.equal(draft.content, '# Timer 1.0\nA timer.\n'); + assert.equal(draft.blob, sub.candidate.prd_blob); + const requirement = await f.call('cto', 'request.get', { request: f.request, path: 'requirements/REQ-POMO-1.json' }); + assert.equal(requirement.content.text, 'A working timer.'); + await assert.rejects(f.call('cto', 'request.get', { ...args, path: 'spec/north-star.md' }), /request-document-refused/); + await assert.rejects(f.call('coder', 'request.get', args), /request-access-refused/); + assert.equal(f.store.get('SELECT count(*) n FROM task_snapshots').n, 0); +}); + +test('completion is acknowledged by the chief delivery effect in the original assistant session', async t => { + const f = await planned(t, true); await advance(f, 'start'); + const commit = 'b'.repeat(40); f.push(commit); const requested = f.review(commit); await advance(f, 'review'); + const v = f.call('reviewer', 'review.verdict', { task_ref: f.task_ref, request: requested.request, verdict: 'approve', body: 'Artifact complete' }); + await advance(f, 'delivery'); f.closeTask(v.verdict); await advance(f, 'complete'); + const intent = intents(f).findLast(e => e.result.intent.step === 'send_message'); + const read = f.call('chief', 'message.receive'); for (const m of read) f.call('chief', 'message.read', { id: m.id }); + await f.bus.requestLifecycle.recover(); + assert.ok(!f.journal.effects('demo', f.request).some(e => e.op === intent.op && e.stage === 'committed')); + const delivered = f.call('chief', 'message.send', { to: 'assistant', in_reply_to: f.source.id, body: `Timer artifact: project timer, ${f.task_ref}, approved verdict ${v.verdict}` }); + await f.bus.requestLifecycle.recover(); + const acknowledged = f.journal.effects('demo', f.request).find(e => e.op === intent.op && e.stage === 'committed'); + assert.equal(acknowledged.result.evidence.effect.message, delivered.id); + assert.ok(f.call('assistant', 'message.receive').some(m => m.id === delivered.id)); +}); + + +test('an opened pre-task selection also resumes durably before a resource plan exists', async t => { + const f = await requestFixture(t); + const source = f.call('human', 'message.send', { to: 'pm', body: 'A second product request' }); + await f.call('pm', 'request.open', { request: source.request }); + await f.bus.requestLifecycle.recover(); + const rows = f.journal.effects('demo', source.request); + const intent = rows.find(e => e.stage === 'uncertain' && e.result?.intent); + assert.equal(intent.result.intent.step, 'intake_propose'); + assert.equal(intent.result.intent.phase, 'selection'); + assert.equal(intent.result.intent.evidence.source_message, source.id); + assert.deepEqual(f.journal.readRequest('demo', source.request).tasks, []); +}); diff --git a/packages/bus/tests/request-fixture.mjs b/packages/bus/tests/request-fixture.mjs new file mode 100644 index 000000000..772f19853 --- /dev/null +++ b/packages/bus/tests/request-fixture.mjs @@ -0,0 +1,56 @@ +import test from 'node:test'; +import { DatabaseSync } from 'node:sqlite'; +import assert from 'node:assert/strict'; +import { mkdirSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { startBroker } from '../src/runtime.mjs'; +import { bootConfig, loadSystem } from '../../cli/src/config.mjs'; +import { fixture, tmp } from '../../cli/tests/helpers.mjs'; +import { prdFixture, git } from '../../project/tests/prd-fixture.mjs'; +import { createRequestStore } from '../src/request-store.mjs'; + +export async function requestFixture(t, mode = 'autonomous', beforeBootstrap = null, threaded = false) { + const root = tmp(t, 'activation-'), f = fixture(root, 'demo', d => { d.launch.instances.push('pm'); if (threaded) d.roles.assistant = { definition: 'assistant' }; return d; }); + const source = join(root, 'project'); mkdirSync(join(source, '.mosaic'), { recursive: true }); + writeFileSync(join(source, '.mosaic/project.json'), JSON.stringify({ projectVersion: 1, id: 'stack', vars: { 'tracker.project': 1 } })); + git(source, 'init', '-qb', 'main'); git(source, 'add', '-A'); git(source, 'commit', '-qm', 'Factory'); + const system = loadSystem({ env: f.env }); + const boot = async () => { + const config = bootConfig({ system, businessId: 'demo', env: f.env, warn: () => {} }); + const bus = await startBroker(config); t.after(() => bus.close().catch(() => {})); return bus; + }; + const bus = await boot(), caps = {}; + const db = new DatabaseSync(join(f.dataRoot, 'bus/bus.sqlite'), { readOnly: true }); t.after(() => db.close()); + const store = { get: (sql, ...a) => db.prepare(sql).get(...a), all: (sql, ...a) => db.prepare(sql).all(...a) }; + const bind = b => { + caps.human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }); + caps.console = b.bindConsole({ business: 'demo', human: 'jason', subject: { provider: 'test', iss: 'test', sub: 'owner' } }); + caps.reader = b.bindReader({ business: 'demo' }); + for (const role of ['pm','cto','coder','reviewer','chief']) { caps[role] = b.bindLaunch({ business: 'demo', role, run: role + '-run', harness: 'pi' }); b.request(caps[role], { verb: 'role.claim' }); } + }; + bind(bus.broker); + const call = (who, verb, args = {}) => bus.broker.request(caps[who], { verb, args }); + await beforeBootstrap?.({ f, bus, caps, call, system }); + let sent; + if (threaded) { + const project = call('human', 'project.create', { name: 'Intake' }).id; + const thread = call('human', 'thread.create', { project, name: 'Timer' }).id; + caps.assistant = bus.broker.bindLaunch({ business: 'demo', role: 'assistant', run: 'assistant-run', harness: 'pi', thread }); + call('assistant', 'role.claim'); + sent = call('human', 'thread.send', { thread, body: 'Build a timer.\n' }); + const relay = call('assistant', 'message.send', { to: 'chief', in_reply_to: sent.id, body: 'Original request ' + sent.request }); + call('chief', 'message.send', { to: 'pm', in_reply_to: relay.id, body: 'Open original request ' + sent.request }); + } else sent = call('human', 'message.send', { to: 'pm', body: 'Build a timer.\n' }); + let state = await call('pm', 'request.open', { request: sent.request }); + state = await call('pm', 'request.propose', { request: sent.request, revision: state.revision, op: 'plan', source_project: 'stack', project: 'timer', name: 'Timer', tracker_project: 2 }); + call('human', 'decision.resolve', { id: state.selection.decision, choice: mode }); + state = await bus.requestLifecycle.resume('demo', sent.request); + const context = { business: { id: 'demo', human: 'jason', projects: { timer: { root: join(f.dataRoot, 'products/demo/timer/code') } } }, project: 'timer', dataRoot: f.dataRoot }; + const p = await prdFixture(t, context); + const args = { request: sent.request, revision: state.revision, op: 'prd1', prd_version: '1.0', records_commit: p.args.records_commit, prd_path: p.args.prd_path, requirement_ids: p.args.requirement_ids }; + const submit = () => call('pm', 'prd.submit', args); + const approve = (submission, who = mode === 'guided' ? 'human' : 'cto') => call(who, 'decision.resolve', { id: submission.decision.id, choice: 'yes' }); + const activate = submission => bus.requirementCatalog.activate({ request: sent.request, decision: submission.decision.id }); + return { ...p, f, bus, system, caps, call, source: sent, state, request: sent.request, args, submit, approve, activate, boot, bind, config: bootConfig({ system, businessId: 'demo', env: f.env, warn: () => {} }), store, journal: createRequestStore(store) }; +} + diff --git a/packages/bus/tests/request-intake.test.mjs b/packages/bus/tests/request-intake.test.mjs index 149aaea82..dd0930565 100644 --- a/packages/bus/tests/request-intake.test.mjs +++ b/packages/bus/tests/request-intake.test.mjs @@ -274,7 +274,7 @@ test('3n_to_3o_is_explicit_offline_preserves_rows_and_old_engine_refuses_new_sch const root = mkdtempSync(join(tmpdir(), 'request-migration-')); t.after(() => rmSync(root, { recursive: true, force: true })); mkdirSync(join(root, 'bus'), { mode: 0o700 }); // Task 2's 3o→3p step widens only request_events.kind, so the 3o schema is the current one with the old list. - const path = join(root, 'bus/bus.sqlite'), currentDDL = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8').replace("'setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','cancelled'", "'setup.wait','cancelled'"); + const path = join(root, 'bus/bus.sqlite'), currentDDL = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8').replace("'setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','advanced','cancelled'", "'setup.wait','cancelled'"); const newDDL = readFileSync(new URL('../src/migrations/3n-3o.sql', import.meta.url), 'utf8'); assert.ok(currentDDL.endsWith(newDDL)); const oldDDL = currentDDL.slice(0, -newDDL.length); const { schemaDigest, SCHEMA_VERSION } = awaitImportStore; @@ -286,7 +286,7 @@ test('3n_to_3o_is_explicit_offline_preserves_rows_and_old_engine_refuses_new_sch db.close(); chmodSync(path, 0o600); const before = readFileSync(path); assert.throws(() => new Store(root), /schema-mismatch/); assert.deepEqual(readFileSync(path), before); const cli = spawnSync(process.execPath, [new URL('../src/migrate-schema.mjs', import.meta.url).pathname, root], { encoding: 'utf8' }); - assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3q ready\n'); assert.equal(SCHEMA_VERSION, '3q'); + assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3r ready\n'); assert.equal(SCHEMA_VERSION, '3r'); const store = new Store(root); assert.equal(store.get("SELECT id FROM events").id, 'preserved'); for (const table of ['request_events','request_effects','request_snapshots','request_snapshot_events','request_project_events']) assert.equal(store.get(`SELECT count(*) n FROM ${table}`).n, 0); store.close(); assert.equal(spawnSync(process.execPath, [new URL('../src/migrate-schema.mjs', import.meta.url).pathname, root]).status, 0); @@ -307,7 +307,7 @@ test('3o_to_3p_widens_only_the_event_kinds_and_keeps_rows_seq_and_guards', t => t.after(() => rmSync(root, { recursive: true, force: true })); mkdirSync(join(root, 'bus'), { mode: 0o700 }); const path = join(root, 'bus/bus.sqlite'); // The 3o schema is the current one with the narrower kind list. - const ddl3o = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8').replace("'setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','cancelled'", "'setup.wait','cancelled'"); + const ddl3o = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8').replace("'setup.wait','bootstrap.stale','bootstrap.completed','prd.submitted','prd.activated','prd.invalidated','mode.changed','advanced','cancelled'", "'setup.wait','cancelled'"); const { schemaDigest } = awaitImportStore, AT = '2026-11-03T00:00:00.000Z'; const db = new DatabaseSync(path); db.exec(ddl3o); db.prepare('INSERT INTO meta VALUES(?,?)').run('schema_version', '3o'); @@ -323,7 +323,7 @@ test('3o_to_3p_widens_only_the_event_kinds_and_keeps_rows_seq_and_guards', t => db.close(); chmodSync(path, 0o600); assert.throws(() => new Store(root), /schema-mismatch/); const cli = spawnSync(process.execPath, [new URL('../src/migrate-schema.mjs', import.meta.url).pathname, root], { encoding: 'utf8' }); - assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3q ready\n'); + assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3r ready\n'); const store = new Store(root); assert.deepEqual(store.all('SELECT * FROM request_events ORDER BY seq'), before, 'every row keeps its seq and bytes'); assert.deepEqual(store.all('SELECT * FROM request_effects ORDER BY seq'), effects); diff --git a/packages/bus/tests/session-clear.test.mjs b/packages/bus/tests/session-clear.test.mjs index b827cb18f..2ebbc065b 100644 --- a/packages/bus/tests/session-clear.test.mjs +++ b/packages/bus/tests/session-clear.test.mjs @@ -530,7 +530,7 @@ test('T4 schema: a 3m store migrates to 3n, session_ops and message_cancels are const before = tables(path); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); const after = tables(path); for (const name of ['session_ops', 'message_cancels', 'request_events', 'request_effects', 'request_snapshots', 'request_snapshot_events', 'request_project_events']) { assert.deepEqual(after[name], [], name); @@ -571,7 +571,7 @@ async function engine3m(root) { const source = readFileSync(here('../src/store.mjs'), 'utf8'); const edited = source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3m).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3m';") + .replace("export const SCHEMA_VERSION = '3r';", "export const SCHEMA_VERSION = '3m';") .replace(/\}, \{\n {2}\/\/ Row 271 \(#1776\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); assert.ok(edited.includes("export const SCHEMA_VERSION = '3m';") && edited.includes('3l-3m') && !edited.includes('3m-3n'), 'the 3m engine has no 3m→3n step'); writeFileSync(join(moduleRoot, 'src/store.mjs'), edited); @@ -588,7 +588,7 @@ test('T4 schema: the 3m engine refuses a 3n store, migrating or not, and leaves const Store3m = await engine3m(root); assert.throws(() => new Store3m(root), /schema-mismatch/); assert.throws(() => new Store3m(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); assert.deepEqual(master(path), was); assert.deepEqual(tables(path), before); // The same engine opens a fresh 3m store, so the refusal is the version's and not a broken copy. diff --git a/packages/bus/tests/sessions.test.mjs b/packages/bus/tests/sessions.test.mjs index b206410ff..cd5a7e83d 100644 --- a/packages/bus/tests/sessions.test.mjs +++ b/packages/bus/tests/sessions.test.mjs @@ -374,7 +374,7 @@ test('T7: a 3k store migrates to 3l, each thread\'s default session has the thre const { root, path } = source3k(t); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); // The migrated schema is the fresh one, object for object. const freshRoot = mkdtempSync(join(tmpdir(), 'sessions-fresh-')); t.after(() => rmSync(freshRoot, { recursive: true, force: true })); @@ -430,7 +430,7 @@ async function engine3k(root) { const source = readFileSync(here('../src/store.mjs'), 'utf8'); const edited = source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3k).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3k';") + .replace("export const SCHEMA_VERSION = '3r';", "export const SCHEMA_VERSION = '3k';") .replace(/\}, \{\n {2}\/\/ Row 264a \(#1769\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); assert.ok(edited.includes("export const SCHEMA_VERSION = '3k';") && edited.includes('3j-3k') && !edited.includes('3k-3l'), 'the 3k engine has no 3k→3l step'); writeFileSync(join(moduleRoot, 'src/store.mjs'), edited); @@ -447,7 +447,7 @@ test('T7 D9: the 3k engine refuses a 3l store, migrating or not, and leaves it a const Store3k = await engine3k(root); assert.throws(() => new Store3k(root), /schema-mismatch/); assert.throws(() => new Store3k(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); assert.deepEqual(master(path), was); assert.deepEqual(rows(path), before); // The same engine opens a fresh 3k store, so the refusal is the version's and not a broken copy. diff --git a/packages/bus/tests/spec-migration.test.mjs b/packages/bus/tests/spec-migration.test.mjs index e6652283c..3ba6b2ec4 100644 --- a/packages/bus/tests/spec-migration.test.mjs +++ b/packages/bus/tests/spec-migration.test.mjs @@ -63,7 +63,7 @@ test('3e copy migrates through 3f to 3g preserving every old column/row and addi assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); try { - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3r'); assert.deepEqual({ ...store.get('SELECT spec_path,spec_commit FROM decisions') }, { spec_path: null, spec_commit: null }); assert.equal(store.get('PRAGMA quick_check').quick_check, 'ok'); assert.deepEqual(store.all('PRAGMA foreign_key_check'), []); diff --git a/packages/bus/tests/thread-migration.test.mjs b/packages/bus/tests/thread-migration.test.mjs index a9bd0d17f..361fefa46 100644 --- a/packages/bus/tests/thread-migration.test.mjs +++ b/packages/bus/tests/thread-migration.test.mjs @@ -49,7 +49,7 @@ async function database3d(t) { const source = readFileSync(here('../src/store.mjs'), 'utf8'); const pinned = source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3d).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3d';") + .replace("export const SCHEMA_VERSION = '3r';", "export const SCHEMA_VERSION = '3d';") .replace(/\}, \{\n {2}from: '3d',[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); assert.ok(!pinned.includes('3d-3e'), 'the 3d engine has no 3d→3e step'); writeFileSync(join(moduleRoot, 'src/store.mjs'), pinned); @@ -101,7 +101,7 @@ test('REQ-THREAD-1 REQ-ORG-2: the 3d→3e migration adds the thread tables and l assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); t.after(() => store.close()); - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3r'); for (const table of ['projects', 'threads', 'thread_claims', 'message_threads', 'thread_opens']) assert.equal(store.get(`SELECT count(*) n FROM ${table}`).n, 0, table); // Parked: the undelivered ones are counted per business and role, and none is changed or given a thread. @@ -161,5 +161,5 @@ test('REQ-THREAD-1: the offline command prints the kept-message counts before th const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname; const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' }); assert.equal(migrated.status, 0, migrated.stderr); - assert.equal(migrated.stdout, [...REPORT, 'bus schema 3q ready', ''].join('\n')); + assert.equal(migrated.stdout, [...REPORT, 'bus schema 3r ready', ''].join('\n')); }); diff --git a/packages/bus/tests/thread-shares.test.mjs b/packages/bus/tests/thread-shares.test.mjs index a5bcd0f48..3f17439ea 100644 --- a/packages/bus/tests/thread-shares.test.mjs +++ b/packages/bus/tests/thread-shares.test.mjs @@ -403,7 +403,7 @@ test('a 3l store migrates to 3m: thread_shares is new and empty, and every other const before = tables(path); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); const after = tables(path); assert.deepEqual(after.thread_shares, []); delete after.thread_shares; @@ -439,7 +439,7 @@ async function engine3l(root) { const source = readFileSync(here('../src/store.mjs'), 'utf8'); const edited = source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3l).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3q';", "export const SCHEMA_VERSION = '3l';") + .replace("export const SCHEMA_VERSION = '3r';", "export const SCHEMA_VERSION = '3l';") .replace(/\}, \{\n {2}\/\/ Row 206 \(193b2[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); assert.ok(edited.includes("export const SCHEMA_VERSION = '3l';") && edited.includes('3k-3l') && !edited.includes('3l-3m'), 'the 3l engine has no 3l→3m step'); writeFileSync(join(moduleRoot, 'src/store.mjs'), edited); @@ -456,7 +456,7 @@ test('the 3l engine refuses a 3m store, migrating or not, and leaves it as it wa const Store3l = await engine3l(root); assert.throws(() => new Store3l(root), /schema-mismatch/); assert.throws(() => new Store3l(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3q'); + assert.equal(version(path), '3r'); assert.deepEqual(master(path), was); assert.deepEqual(tables(path), before); // The same engine opens a fresh 3l store, so the refusal is the version's and not a broken copy. diff --git a/packages/bus/tests/usage-migration.test.mjs b/packages/bus/tests/usage-migration.test.mjs index b42e4f73f..7a4bc91cb 100644 --- a/packages/bus/tests/usage-migration.test.mjs +++ b/packages/bus/tests/usage-migration.test.mjs @@ -68,7 +68,7 @@ test('REQ-LIMIT-4 3g copy migrates through 3h, 3i, 3j and 3k to 3l preserving ev assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); try { - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3q'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3r'); assert.deepEqual({ ...store.get('SELECT to_name FROM messages') }, { to_name: null }); assert.equal(store.get('SELECT count(*) AS n FROM usage_records').n, 0); assert.equal(store.get('SELECT name FROM instance_names').name, 'Builder'); diff --git a/packages/cli/src/launcher.mjs b/packages/cli/src/launcher.mjs index ec602ce4c..c602fe1c8 100644 --- a/packages/cli/src/launcher.mjs +++ b/packages/cli/src/launcher.mjs @@ -292,6 +292,7 @@ export function createLauncher({ const refusedRelease = new Map(); let refusedHolders = null; let refusedRequests = null; + const requestDrains = new Map(); const sockets = new Set(); const gitSocket = createGitSocket({ host, business: b, dataRoot, env, log, ...git }); @@ -796,12 +797,57 @@ export function createLauncher({ await threadPass(); await landPass(); } + await requestPass(halted); await lifecyclePass(); } finally { ticking = false; } } + // Request recovery starts only the configured chief through the existing --pm path. + // Workers always come back through that chief's ordinary launch tool/capability. + async function requestPass(halted = false) { + for (const [run, request] of launches && !halted ? requestDrains : []) { + const session = registry.sessions.find(s => s.run === run); + if (!session) { requestDrains.delete(run); continue; } + try { + const out = await host.op({ op: 'request.release', cap: request.cap, record: { business: b.id, run, instance: session.instance, project: request.project } }); + if (out?.released) { terminate([session]); requestDrains.delete(run); } + } catch (e) { + // A revoked chief capability cannot keep retiring workers. + requestDrains.delete(run); + await host.op({ op: 'request.launchResult', record: { business: b.id, instance: session.instance, code: e.code ?? 'launch-failed', run: null } }); + } + } + const pending = await host.op({ op: 'request.launches', business: b.id }); + if (!Array.isArray(pending) || !pending.length) return; + if (!launches || halted) { + await host.op({ op: 'request.launchResult', record: { business: b.id, instance: b.launch.by, code: !launches ? 'launches-disabled' : 'processing-halted', run: null } }); + return; + } + for (const i of pending) { + const session = registry.running(b.id, i.target, null); + if (!session || !i.project || stopping.has(session.run)) continue; + let projects = []; + try { projects = JSON.parse(readFileSync(join(session.runDir, 'bundle/manifest.json'), 'utf8')).projects ?? []; } catch {} + if (!projects.includes(i.project)) { + const ready = await host.op({ op: 'request.admission', business: b.id, project: i.project }); + if (ready?.ready === true) await host.op({ op: 'request.manifest', record: { business: b.id, instance: i.target, project: i.project, run: session.run } }); + } + } + const chief = b.launch?.by; + if (!chief || registry.running(b.id, chief, null) || pending.some(i => i.chief_run)) return; + const next = pending.find(i => i.chief === chief && i.chief_attempt !== i.prerequisite); + if (!next) return; + try { + await host.op({ op: 'request.chief', business: b.id }); + const out = await launchPm(); + await host.op({ op: 'request.launchResult', record: { business: b.id, instance: chief, code: null, run: out.run } }); + } catch (e) { + await host.op({ op: 'request.launchResult', record: { business: b.id, instance: chief, code: e.code ?? 'launch-failed', run: null } }); + } + } + async function threadPass() { let threads; try { @@ -1237,6 +1283,34 @@ export function createLauncher({ record({ event: "refused", by: who.role, instance, code: "launches-disabled" }); return { ok: false, error: "launches-disabled" }; } + const reply = async out => { + if (typeof instance === 'string') await host.op({ op: 'request.launchResult', record: { business: b.id, instance, code: out.ok ? null : out.error, run: out.ok ? out.result.run : null } }); + return out; + }; + try { b = loadBusiness(host.business, { dir: configDir(env), rolesDir: rolesDir(env) }); } + catch { return reply({ ok: false, error: 'organization-refused' }); } + // Product enums and clones are frozen in a session's manifest. A legitimate chief + // launch may request a drain, never interrupt the turn or start a privileged worker. + const running = registry.running(b.id, instance, null); + if (running && who.role === b.launch?.by) { + const pending = await host.op({ op: 'request.launches', business: b.id }); + const needed = pending.filter(i => i.target === instance && i.project); + let projects = []; + try { projects = JSON.parse(readFileSync(join(running.runDir, 'bundle/manifest.json'), 'utf8')).projects ?? []; } catch {} + const missing = needed.find(i => !projects.includes(i.project)); + if (missing) { + if (!b.launch?.instances.includes(instance)) return reply({ ok: false, error: 'instance-not-listed' }); + try { + await host.op({ op: 'authorizeLaunch', cap, instance }); + const ready = await host.op({ op: 'request.admission', business: b.id, project: missing.project }); + if (ready?.ready !== true) return reply({ ok: false, error: ready?.code ?? 'project-pending' }); + const out = await host.op({ op: 'request.release', cap, record: { business: b.id, run: running.run, instance, project: missing.project } }); + if (out?.released) terminate([running]); + else requestDrains.set(running.run, { cap, project: missing.project }); + return reply({ ok: false, error: 'request-manifest-draining' }); + } catch (e) { return reply({ ok: false, error: e.code ?? 'launch-failed' }); } + } + } let checked; try { if (typeof instance !== "string") throw new Refusal("invalid-request"); @@ -1245,22 +1319,22 @@ export function createLauncher({ if (!(e instanceof Refusal)) throw e; await host.op({ op: "refuse", cap, code: e.code }).catch(() => {}); record({ event: "refused", by: who.role, instance, code: e.code, why: e.message, ...(e.limit?{limit:e.limit}:{}) }); - return { ok: false, error: e.code }; + return reply({ ok: false, error: e.code }); } try { await host.op({ op: "authorizeLaunch", cap, instance }); } catch (e) { record({ event: "refused", by: who.role, instance, code: e.code ?? "not-authorized" }); - return { ok: false, error: e.code ?? "not-authorized" }; + return reply({ ok: false, error: e.code ?? "not-authorized" }); } try { - return { ok: true, result: await start({ instance, launchedBy: who.role, ...checked }) }; + return reply({ ok: true, result: await start({ instance, launchedBy: who.role, ...checked }) }); } catch (e) { const code = e instanceof Refusal ? e.code : "launch-failed"; log(`launch of ${instance} for ${who.role} failed: ${e.message}`); await host.op({ op: "refuse", cap, code }).catch(() => {}); record({ event: "refused", by: who.role, instance, code, why: e.message, ...(e.limit?{limit:e.limit}:{}) }); - return { ok: false, error: code }; + return reply({ ok: false, error: code }); } }); } diff --git a/packages/cli/tests/request-launch.test.mjs b/packages/cli/tests/request-launch.test.mjs new file mode 100644 index 000000000..3cb8cfc28 --- /dev/null +++ b/packages/cli/tests/request-launch.test.mjs @@ -0,0 +1,182 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { spawnSync } from 'node:child_process'; +import { readFileSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { fixture, tmp, withoutHostCredentials, fakeHarnesses } from './helpers.mjs'; +import { bootConfig, loadSystem } from '../src/config.mjs'; +import { startHost, startTimeOf } from '../src/host.mjs'; +import { createLauncher } from '../src/launcher.mjs'; +import { Client } from '../../bus/src/client.mjs'; +import { gitEnv } from '../../seat/src/checkout.mjs'; + +async function setup(t, edit = d => d, { revoked = false } = {}) { + const root = tmp(t, 'request-launch-'), f = fixture(root, 'demo', edit); + withoutHostCredentials(f.env, { piKey: true }); + const repo = f.doc.projects.stack.root; + for (const args of [['init','-qb','main'],['-c','user.name=test','-c','user.email=test@example.invalid','commit','--allow-empty','-qm','seed']]) { + const r = spawnSync('git', ['-C', repo, ...args], { env: gitEnv(), encoding: 'utf8' }); assert.equal(r.status, 0, r.stderr); + } + const fake = fileURLToPath(new URL('../../harness/tests/fixtures/fake-adapter.mjs', import.meta.url)); + const adapter = join(root, 'adapter.sh'); writeFileSync(adapter, `exec '${process.execPath}' '${fake}'\n`); + const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter }); + const system = loadSystem({ env: f.env }), host = await startHost({ boot: bootConfig({ system, businessId: 'demo', env: f.env }), business: 'demo' }); + const launcher = createLauncher({ host, system, env: f.env, makeLoader, harnessVersion: 'fixture', readOnly: [fake], sessionDefaults: { pollInterval: 30 } }); + await launcher.listen(); + t.after(() => host.close(0)); + if (revoked) { + const { DatabaseSync } = await import('node:sqlite'); + const db = new DatabaseSync(join(f.dataRoot, 'bus/bus.sqlite')); + db.prepare("INSERT INTO events(id,at,business,kind,subject,body) VALUES('fixture-revoked',?,'demo','launch.revoked',NULL,'{}')").run(new Date().toISOString()); + db.close(); + } + const bound = {}; + for (const role of ['chief','pm']) { + bound[role] = await host.bindLaunch({ business: 'demo', role, run: role+'-fixture', harness: 'pi', pid: process.pid, startTime: startTimeOf(process.pid) }); + const c = new Client({ path: host.path, cap: bound[role].cap }); await c.call('role.claim'); + } + return { root, f, host, launcher, bound }; +} + +test('PM cannot use continuation launch recovery to gain role.launch', async t => { + const x = await setup(t); + const out = await x.launcher.fromSession(x.bound.pm.cap, 'coder'); + assert.equal(out.ok, false); + assert.ok(['decision-required','not-authorized'].includes(out.error), out.error); +}); + +test('chief launch refusals are returned unchanged and stored through the host request path', async t => { + const x = await setup(t); + const c = new Client({ path: x.host.path, cap: x.bound.pm.cap }); + // Host operations are not broker verbs and cannot be invoked by a managed session. + await assert.rejects(c.call('request.launchResult', { instance: 'coder', code: 'capacity-full' }), /unknown-verb/); + const out = await x.launcher.fromSession(x.bound.chief.cap, 'absent'); + assert.equal(out.error, 'instance-not-listed'); + const pending = await x.host.op({ op: 'request.launches', business: 'demo' }); + assert.deepEqual(pending, []); +}); + +async function until(fn) { + const end = Date.now() + 10000; + while (Date.now() < end) { const value = await fn(); if (value) return value; await new Promise(r => setTimeout(r, 25)); } + throw Error('request launch did not settle'); +} + +test('the real chief launches an existing worker with the admitted project in its manifest', async t => { + const x = await setup(t); + const out = await x.launcher.fromSession(x.bound.chief.cap, 'coder'); + assert.equal(out.ok, true, out.error); + const session = x.launcher.registry.sessions.find(s => s.run === out.result.run); + const manifest = JSON.parse(readFileSync(join(session.runDir, 'bundle/manifest.json'))); + assert.deepEqual(manifest.projects, ['stack']); + assert.equal(manifest.definition, 'coder'); + assert.ok(manifest.typed.includes('push_working')); + await x.host.close(0); +}); + +test('revoked launch and full model limits preserve their existing refusal codes', async t => { + for (const code of ['launch-revoked','capacity-full']) await t.test(code, async t => { + const x = await setup(t, doc => { + if (code === 'capacity-full') doc.launch.limits.models.zai['glm-5.3-flash'] = 0; + return doc; + }, { revoked: code === 'launch-revoked' }); + const out = await x.launcher.fromSession(x.bound.chief.cap, 'coder'); + assert.equal(out.error, code); + assert.equal(x.launcher.registry.sessions.length, 0); + }); +}); + +function requestHost(x) { + const operations = { + identity: m => x.bus.identity(m.cap), authorizeLaunch: m => x.bus.authorizeLaunch(m.cap, m.instance), + refuse: m => x.bus.refuse(m.cap, m.code), credentialStatus: m => x.bus.credentialStatus(m.business, m.role), + modelPreflight: m => x.bus.modelPreflight(m), mayLaunch: m => x.bus.mayLaunch(m), + landHeld: m => x.bus.landHeld(m.record), landHolders: m => x.bus.landHolders(m.business), + sessionClears: m => x.bus.sessionClears(m.business), + threadsState: m => x.bus.threadsState(m.business), lifecycleRequests: m => x.bus.lifecycleRequests(m.business), + 'request.admission': m => x.bus.admission(m), 'request.launches': m => x.bus.requestLaunches(m.business), + 'request.launchResult': m => x.bus.requestLaunchResult(m.record), 'request.manifest': m => x.bus.requestManifest(m.record), + 'request.release': m => x.bus.requestRelease(m.cap, m.record), 'request.chief': m => x.bus.requestChief(m.business), + endLaunch: m => x.bus.endLaunch(m.record), + }; + return { business: 'demo', path: x.bus.path, dataRoot: x.f.dataRoot, beforeClose: () => {}, + op: m => Promise.resolve().then(() => { + assert.ok(operations[m.op], `unexpected host operation ${m.op}`); return operations[m.op](m); + }), + bindLaunch: record => { const out = x.bus.bindLaunch(record); x.caps[record.role] = out.cap; return out; } }; +} + +test('an old product manifest drains its active turn before the chief relaunches through normal admission', { timeout: 30000 }, async t => { + const { requestFixture } = await import('../../bus/tests/request-fixture.mjs'); + let launcher, worker, host, entered, released; + const f = await requestFixture(t, 'autonomous', async x => { + withoutHostCredentials(x.f.env, { piKey: true }); + const adapter = join(x.f.dataRoot, 'hold.sh'), source = join(x.f.dataRoot, 'hold.mjs'); + writeFileSync(source, `import {writeFileSync,existsSync} from 'node:fs'; +const marker=process.env.MOSAIC_TURN_MARKER; +if ((process.env.MOSAIC_REQUEST??'').includes('HOLD-REQUEST-TURN')) { + writeFileSync(marker+'.entered','entered'); + while(!existsSync(marker+'.release')) await new Promise(r=>setTimeout(r,25)); +} +writeFileSync(marker,'done'); process.stdout.write('turn completed');`); + writeFileSync(adapter, `exec '${process.execPath}' '${source}'\n`); + const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter }); + host = requestHost(x); + launcher = createLauncher({ host, system: x.system, env: x.f.env, makeLoader, harnessVersion: 'fixture', readOnly: [source], sessionDefaults: { pollInterval: 25 } }); + t.after(() => launcher.close()); + x.bus.broker.endLaunch({ business: 'demo', run: 'pm-run', reason: 'exit', exitCode: 0 }); + const out = await launcher.fromSession(x.caps.chief, 'pm'); assert.equal(out.ok, true, out.error); + worker = launcher.registry.sessions.find(s => s.run === out.result.run); + entered = join(worker.runDir, 'turn.done.entered'); released = join(worker.runDir, 'turn.done.release'); + await until(async () => { try { x.call('pm', 'state'); return true; } catch { return false; } }); + x.call('human', 'message.send', { to: 'pm', body: 'HOLD-REQUEST-TURN' }); + const { existsSync } = await import('node:fs'); await until(() => existsSync(entered)); + }); + const original = JSON.parse(readFileSync(join(worker.runDir, 'bundle/manifest.json'))); + assert.deepEqual(original.projects, ['stack']); + await f.bus.requestLifecycle.recover(); + await launcher.tick(); + const chief = f.call('chief', 'message.receive'); + assert.ok(chief.some(m => m.body.includes('lacks admitted project timer'))); + const refused = await launcher.fromSession(f.caps.chief, 'pm'); + assert.equal(refused.error, 'request-manifest-draining'); + assert.ok(launcher.registry.sessions.some(s => s.run === worker.run), 'active turn remains alive'); + assert.equal((await f.call('human', 'request.get', { request: f.request })).launch_refusal.code, 'request-manifest-draining'); + writeFileSync(released, 'release'); + // The runtime keeps the same drain intent while the turn finishes; the chief path owns the release. + await until(async () => { await launcher.tick(); return !launcher.registry.sessions.some(s => s.run === worker.run); }); + const next = await launcher.fromSession(f.caps.chief, 'pm'); assert.equal(next.ok, true, next.error); + const successor = launcher.registry.sessions.find(s => s.run === next.result.run); + assert.deepEqual(JSON.parse(readFileSync(join(successor.runDir, 'bundle/manifest.json'))).projects, ['stack','timer']); + const { existsSync } = await import('node:fs'); + assert.equal(existsSync(join(f.f.dataRoot, 'workspaces/demo/pm/timer/.git')), true); + await launcher.close(); +}); + +test('an unavailable chief records its exact refusal, stays quiet, and retries the same path on admission revision change', { timeout: 30000 }, async t => { + const { requestFixture } = await import('../../bus/tests/request-fixture.mjs'); + const f = await requestFixture(t); withoutHostCredentials(f.f.env, { piKey: true }); + const { dirname } = await import('node:path'); + const config = join(dirname(f.f.config), 'businesses/demo.json'); + const doc = JSON.parse(readFileSync(config)); + doc.launch.limits.models.zai['glm-5.3-flash'] = 0; writeFileSync(config, JSON.stringify(doc)); + f.bus.broker.endLaunch({ business: 'demo', run: 'chief-run', reason: 'exit', exitCode: 0 }); + const fake = fileURLToPath(new URL('../../harness/tests/fixtures/fake-adapter.mjs', import.meta.url)); + const adapter = join(f.f.dataRoot, 'chief.sh'); writeFileSync(adapter, `exec '${process.execPath}' '${fake}'\n`); + const makeLoader = await fakeHarnesses(t, { pi: adapter, 'claude-code': adapter }); + const launcher = createLauncher({ host: requestHost({ bus: f.bus, f: f.f, caps: f.caps }), system: f.system, env: f.f.env, makeLoader, harnessVersion: 'fixture', readOnly: [fake], sessionDefaults: { pollInterval: 25 } }); + t.after(() => launcher.close()); + await launcher.listen(); + await f.bus.requestLifecycle.recover(); await launcher.tick(); + assert.equal((await f.call('pm', 'request.get', { request: f.request })).launch_refusal.code, 'capacity-full'); + const attempts = f.journal.effects('demo', f.request).filter(e => e.result?.type === 'launch').length; + for (let n = 0; n < 3; n++) await launcher.tick(); + assert.equal(f.journal.effects('demo', f.request).filter(e => e.result?.type === 'launch').length, attempts); + doc.launch.limits.models.zai['glm-5.3-flash'] = 1; writeFileSync(config, JSON.stringify(doc)); + await launcher.tick(); + assert.equal(launcher.registry.sessions.length, 1, JSON.stringify(await f.call('pm', 'request.get', { request: f.request }))); + assert.equal(launcher.registry.sessions[0].instance, 'chief'); + assert.equal((await f.call('pm', 'request.get', { request: f.request })).launch_refusal, null); + await launcher.close(); +}); diff --git a/packages/harness/src/bundle.mjs b/packages/harness/src/bundle.mjs index 8199cd316..f4c65b148 100644 --- a/packages/harness/src/bundle.mjs +++ b/packages/harness/src/bundle.mjs @@ -381,6 +381,7 @@ export function buildBundle({ dir, resolved, contract, arbiter = false, tracker harnessVersion, builtins, typed: tools.map((t) => harness.typedName(t.name)), + projects: (session.checkouts ?? []).map(c => c.project), untyped, webSource: resolved.limits.tools.includes("web") ? web.source : null, skills: skills.map(({ name, path, sha256, files }) => ({ name, path, sha256, files })), diff --git a/packages/harness/src/tools.mjs b/packages/harness/src/tools.mjs index 1e4262df8..267108940 100644 --- a/packages/harness/src/tools.mjs +++ b/packages/harness/src/tools.mjs @@ -55,6 +55,8 @@ const TASK_TOOLS = { labels: LABEL_IDS, relations: { type: "array", items: RELATION, maxItems: 50 }, request: str("The request id of the human message this task answers, shown as \"request \" in that message's header; not the message's own id"), + project_key: str("Admitted product project key"), + requirement_snapshot: str("Exact approved requirement snapshot id from list_requirements"), requirement: str("Requirement id from list_requirements, such as REQ-CRED-2", { pattern: "^REQ-[A-Z]+-[1-9][0-9]*$" }), decision: COMMON.decision, }, ["title", "request", "requirement"]], @@ -117,9 +119,9 @@ export function typedTools({ resolved, arbiter = false, tracker = false, project }, ["roles"])); } // Intake precedes tasks, tracker availability and project checkouts. Tool presence does not - // widen the broker's request ownership checks (currently PM/owner, including for reads). + // widen the broker's request ownership checks (PM/owner and the eligible technical arbiter for reads). if (["pm", "cto"].includes(resolved.definition)) { - add("intake_status", "request.get", null, "Read the durable intake state for a request. The broker checks request access on every call.", obj({ request: REQUEST }, ["request"])); + add("intake_status", "request.get", null, "Read the durable intake state for a request. The broker checks request access on every call. Pass path to read one exact selected PRD or requirement document from the candidate.", obj({ request: REQUEST, path: str("Optional exact candidate prd_path or requirement path from intake_status") }, ["request"])); add("intake_projects", "request.projects", null, "List the accessible existing software project mappings for a request. The broker checks request access and filters the projects.", obj({ request: REQUEST }, ["request"])); } if (resolved.definition === "pm" && held.has("message.send")) { @@ -134,6 +136,16 @@ export function typedTools({ resolved, arbiter = false, tracker = false, project tracker_project: { type: "integer", minimum: 1, description: "Installed tracker board id to display in the resource proposal" }, }, ["request", "revision", "op", "source_project", "project", "name", "tracker_project"])); } + if (resolved.definition === "pm") { + const control = { request: REQUEST, revision: { type: "integer", minimum: 1 }, op: str("Stable operation id; reuse only for the identical retry") }; + if (held.has("message.send")) add("advance_request", "request.advance", "message.send", "Verify evidence for the next fixed request transition. Never claim a success state; the broker derives it from admitted tasks, pushes, reviews and closure receipts.", obj(control, ["request", "revision", "op"])); + if (held.has("record.write")) add("submit_prd", "prd.submit", "record.write", "Submit exact committed discovery records for PRD approval. The broker reads and binds every blob, author and requirement; submission grants no approval.", obj({ ...control, + prd_version: str("Product PRD version, such as 1.0", { pattern: "^[0-9]+\\.[0-9]+(?:\\.[0-9]+)?$" }), + records_commit: str("Exact records repository commit", { pattern: "^[a-f0-9]{40}([a-f0-9]{24})?$" }), + prd_path: str("docs/prd-VERSION.md"), + requirement_ids: { type: "array", minItems: 1, maxItems: 64, uniqueItems: true, items: { type: "string" } }, + }, ["request", "revision", "op", "prd_version", "records_commit", "prd_path", "requirement_ids"])); + } // Row 194 (#1692, D3 item 5): the instance is a name, not an enum fixed when the bundle was built, // so the chief launches its own hire without a restart. The host checks the listing at each launch. if (resolved.launch) { @@ -338,7 +350,7 @@ export function typedTools({ resolved, arbiter = false, tracker = false, project add("get_task", "tasks", null, "Read one tracker task in full, description included.", obj({ task_ref: TASK_REF }, ["task_ref"])); // Row 90 (#1584): the ids task_create takes, for the sessions that hold it. if (held.has("task.create")) - add("list_requirements", "requirements", null, "List the approved PRD's requirement ids, each with a one-line title. task_create takes only these. When none fits a request, raise a prd.approve decision instead of creating the task.", obj({})); + add("list_requirements", "requirements", null, "List the approved PRD's requirement ids, each with a one-line title. task_create takes only these. When none fits a request, raise a prd.approve decision instead of creating the task.", obj({ project_key: str("Admitted product project key"), requirement_snapshot: str("Exact approved snapshot; omit to read the current admission") })); } // mos-1308 / M3: no generic spec, evidence, tracker, decision or write tools. The // C2-08 web grant still supplies broker web reads to companion and research only. diff --git a/packages/harness/tests/bundle.test.mjs b/packages/harness/tests/bundle.test.mjs index 73323dbba..2bbc88eac 100644 --- a/packages/harness/tests/bundle.test.mjs +++ b/packages/harness/tests/bundle.test.mjs @@ -60,7 +60,7 @@ test("a pi bundle: prompt, policy, tools and manifest, 0600 in a 0700 directory" const prompt = readFileSync(b.files.prompt, "utf8"); assert.match(prompt, /^# Contract\n\nDo the work\.\n\n## This session\n/); assert.match(prompt, /Role instance: pm \(definition pm\)\. Run: run-1\./); - assert.match(prompt, /Bus tools: send_message, request_staff, intake_status, intake_projects, intake_open, intake_propose, task_create, /); + assert.match(prompt, /Bus tools: send_message, request_staff, intake_status, intake_projects, intake_open, intake_propose, advance_request, submit_prd, task_create, /); assert.doesNotMatch(prompt, /Bus tools: [^.]*\blaunch\b/); // Row 90 (#1584): PM fault 10 (#228 c91). assert.match(prompt, /^- Only the human stops a run, with `mosaic stop`\. If you think a run should stop, raise a role\.revoke decision with blocking false/m); diff --git a/packages/harness/tests/request-tools.test.mjs b/packages/harness/tests/request-tools.test.mjs index f97f9a580..39d789668 100644 --- a/packages/harness/tests/request-tools.test.mjs +++ b/packages/harness/tests/request-tools.test.mjs @@ -182,14 +182,16 @@ test('request_staff rejects both/neither references, invalid or unreachable requ assert.equal(x.store.get("SELECT count(*) n FROM messages WHERE class='STAFF-REQUEST'").n, 0); }); -test('broker keeps wrong-role intake refusals, including the existing PM-only read boundary', async t => { +test('broker keeps wrong-role intake refusals, with eligible CTO reads', async t => { const x = await setup(t), args = { request: x.source.request }; const opened = await x.tool('intake_open', args); for (const role of ['cto', 'coder', 'reviewer', 'assistant', 'chief']) { - for (const verb of ['request.open', 'request.get', 'request.projects']) + for (const verb of role === 'cto' ? ['request.open'] : ['request.open', 'request.get', 'request.projects']) await assert.rejects(async () => x.call(role, verb, args), /request-access-refused/); await assert.rejects(async () => x.call(role, 'request.propose', { ...args, revision: opened.revision, op: 'bad-role', source_project: 'factory', project: 'timer', name: 'Timer', tracker_project: 2 }), /request-access-refused/); } + assert.equal((await x.call('cto', 'request.get', args)).request, x.source.request); + assert.equal((await x.call('cto', 'request.projects', args))[0].project, 'factory'); await assert.rejects(x.tool('intake_open', { request: x.source.id }), /request-origin-refused/); await assert.rejects(x.tool('intake_status', { ...args, extra: true }), /invalid-request/); await assert.rejects(x.tool('intake_propose', { ...args, revision: 0, op: 'bad-revision', source_project: 'factory', project: 'timer', name: 'Timer', tracker_project: 2 }), /invalid-request/); @@ -207,3 +209,26 @@ test('assistant cannot coordinate directly with PM or reviewer even with a reque for (const to of ['pm', 'reviewer']) assert.throws(() => x.call('assistant', 'message.send', { to, body: `Request ${x.source.request}`, in_reply_to: x.source.id }), /message-recipient-refused/); }); + +for (const harness of ['pi', 'claude-code']) test(`${harness}: PM submit and advance are typed without a tracker; others cannot write`, async t => { + const x = await setup(t, harness); + const invoke = harness === 'claude-code' ? mcp(t, x.pm.files.tools, x.socket) : x.tool; + const state = await invoke('intake_open', { request: x.source.request }); + await assert.rejects(invoke('advance_request', { request: state.request, revision: state.revision, op: 'advance1', phase: 'complete' }), /invalid-request/); + await assert.rejects(invoke('advance_request', { request: state.request, revision: state.revision, op: 'advance1' }), /request-evidence-required/); + await assert.rejects(invoke('submit_prd', { request: state.request, revision: state.revision, op: 'prd1', prd_version: '1.0', records_commit: 'a'.repeat(40), prd_path: 'docs/prd-1.0.md', requirement_ids: ['REQ-TIMER-1'] }), /request-phase-refused/); + for (const role of ['cto','coder','reviewer','assistant']) { + const b = bundle(t, role, harness); + assert.ok(!b.tools.some(t => ['submit_prd','advance_request'].includes(t.name))); + } +}); + +test('product task and catalog tools carry project and exact snapshot fields', t => { + const resolved = resolvedFor(t, 'pm'); + const tools = typedTools({ resolved, projects: ['timer'], tracker: true }); + for (const name of ['task_create', 'list_requirements']) { + const schema = tools.find(t => t.name === name).parameters; + assert.equal(schema.properties.project_key.type, 'string'); + assert.equal(schema.properties.requirement_snapshot.type, 'string'); + } +}); diff --git a/packages/harness/tests/tools.test.mjs b/packages/harness/tests/tools.test.mjs index 48a48df6b..630481c78 100644 --- a/packages/harness/tests/tools.test.mjs +++ b/packages/harness/tests/tools.test.mjs @@ -15,7 +15,7 @@ test("the PM gets its task verbs and the reads, and no launch", (t) => { const tools = typedTools({ resolved: pm, arbiter: true, tracker: true }); assert.deepEqual(names(tools), [ "send_message", "request_staff", - "intake_status", "intake_projects", "intake_open", "intake_propose", + "intake_status", "intake_projects", "intake_open", "intake_propose", "advance_request", "submit_prd", "task_create", "task_assign", "task_reassign", "task_schedule", "task_priority_change", "task_scope_change", "task_update_assigned", "task_close", "read_spec", "write_record", "raise_decision", "resolve_decision", "list_agents", "inbox", "trail", "list_tasks", "get_task", "list_requirements", @@ -232,7 +232,7 @@ test("list_requirements reads the requirement ids, for a task.create holder with const tools = typedTools({ resolved: pm, arbiter: true, tracker: true }); const list = tools.find((x) => x.name === "list_requirements"); assert.deepEqual([list.verb, list.action], ["requirements", null]); - assert.deepEqual(list.parameters.properties, {}); + assert.deepEqual(Object.keys(list.parameters.properties), ["project_key", "requirement_snapshot"]); assert.match(list.description, /task_create takes only these/); assert.match(list.description, /raise a prd\.approve decision instead of creating the task/); assert.ok(!names(typedTools({ resolved: pm, arbiter: true })).includes("list_requirements")); diff --git a/roles/assistant.md b/roles/assistant.md index 541291edd..cddb3ff17 100644 --- a/roles/assistant.md +++ b/roles/assistant.md @@ -8,6 +8,13 @@ sessions. ## Duties +- Preserve the original user request through assistant → chief → PM intake. + Intake can precede every task and tracker record. Do not coordinate directly + with PM or reviewer. When the chief delivers completion, pass the concrete + artifact locations and task/review evidence to the user in the originating + conversation. A repeated continuation key asks for receipt-based recovery; + never present a read notice, launch, or unverified phase claim as completion. + - Talk with the user. Answer what they ask, say plainly what you don't know and keep their requests in their own words. - Send each request for the business to the chief of staff as a message, diff --git a/roles/chief-of-staff.md b/roles/chief-of-staff.md index dff66408b..8f64247c7 100644 --- a/roles/chief-of-staff.md +++ b/roles/chief-of-staff.md @@ -9,6 +9,19 @@ product's code, you don't review it and you don't create tasks yourself. ## Duties +- Intake staffing may cite the original request before any task exists. Keep + assistant → chief → PM intake and chief → assistant artifact delivery intact. + A durable continuation names its request, revision, phase, evidence and next + typed operation. Launch existing configured PM/CTO/coder/reviewer instances + through `launch` under the current grants and limits. Do not hire them through + your empty hireable list, relay a user command, or ask PM to launch. +- Preserve every exact launch refusal in your reply. The host records it for + intake status and retries the same chief path after prerequisites change. + When a role's frozen manifest lacks an admitted product, request its ordinary + launch: the host waits for the current turn to drain before ending it. Do not + interrupt unrelated active work. A repeated continuation key means resume and + check the existing receipts, not repeat an effect already completed. + - Take the work the assistant passes you. Send it to the role that does it, addressed by role, in the words the assistant gave you. The `pm` role turns product work into tasks and decides who holds them and when diff --git a/roles/coder.md b/roles/coder.md index 10d93e923..370b05e02 100644 --- a/roles/coder.md +++ b/roles/coder.md @@ -5,6 +5,15 @@ assigned to you and hand finished work to review. ## Duties +- A request continuation carries the source request, revision, approved + snapshot, tasks and evidence. Check the current assigned tasks and existing + push/create receipts before resuming; the same key may arrive in a successor + after a dead run. Implement the admitted requirements, finish the artifacts, + push through `push_working` and ask for review of the exact latest commit. + Do not treat message delivery/read as completion or claim a lifecycle phase. + If your frozen tool manifest lacks the product, report it through the chief + path and finish the current turn before its normal relaunch. + - Work only on tasks assigned to you. Update a task's state and comments as the work moves. - Push only with `push_working`. It sends one branch of your clone to diff --git a/roles/cto.md b/roles/cto.md index 4911db8b5..debd60867 100644 --- a/roles/cto.md +++ b/roles/cto.md @@ -6,6 +6,16 @@ assigned engineering work yourself. ## Duties +- Before tasks exist, use `intake_status` and `intake_projects` for requests + in your technical arbiter scope. Help PM discover and commit concrete product + requirements without inventing task evidence. For an autonomous PRD decision, + inspect the exact candidate, records commit and requirement blobs with + `intake_status` path reads, one selected document at a time; approve + only as a distinct reviewer, or request changes. Guided approval remains the + owner's. A continuation's request/revision and evidence identify work to + resume; inspect existing effects before retrying. Reading it acknowledges no + product outcome. Never claim a lifecycle transition for PM. + - Decide how the system is built: design, interfaces, data shapes and which risks are acceptable. Write your decisions down where the next session can find them. diff --git a/roles/pm.md b/roles/pm.md index 987d148c1..2ef98558f 100644 --- a/roles/pm.md +++ b/roles/pm.md @@ -6,6 +6,21 @@ order. You don't write the product's code and you don't review it. ## Duties +- Resume a request continuation by its source request, revision and evidence ids. + Use `intake_status` before acting and check existing effect/task receipts before + retries. A read notice is not completion. During discovery, read the committed + spec and write the product PRD and requirement records with `write_record`. + Submit their exact version, commit, path and ids through `submit_prd`; never + approve your own draft. Autonomous technical approval goes to the distinct CTO; + guided approval stays with the owner. No task exists before snapshot activation. +- After activation, use scoped `list_requirements` and `task_create` with the + product key and exact requirement snapshot. Assign the tasks, request coder + and reviewer staff from the chief, and call `advance_request` with the current + revision and a stable operation id. It checks real task, push, review and + closure evidence for the next fixed transition; it accepts no claimed phase. + Finish the artifacts and close each task against its latest approve verdict, + then advance to delivery through the chief and assistant. + - For a new-project request passed through the assistant and chief of staff, use `intake_open` with the original `request ` from the user's message, then `intake_status` to check its phase, revision and wait. Intake needs no diff --git a/roles/reviewer.md b/roles/reviewer.md index 57c485c23..a6d241324 100644 --- a/roles/reviewer.md +++ b/roles/reviewer.md @@ -5,6 +5,14 @@ candidates and record a verdict. You don't change the code you review. ## Duties +- Resume a request continuation from its source request, revision, tasks, + snapshot and evidence ids. Check existing review verdicts before repeating + work. Fetch the current review request with `fetch_review`, inspect the + completed artifacts and return `review_verdict` for that exact request and + latest commit. Reading a continuation is not completion. Report a missing + product manifest through the chief path and let your current turn drain + before its normal relaunch; do not interrupt other work. + - Review the candidate named in the request against the task's acceptance criteria. A REVIEW-REQUEST names a request id: call `fetch_review` with it, and read and test that commit in your own