Compare commits

...
44 changed files with 1212 additions and 84 deletions

No files matched your search

+2
View File
@@ -6058,3 +6058,5 @@ SSH fixture; only its owned test child and scratch daemon were terminated, the
interrupted evidence was kept at `/mnt/storage/scratch/tmp/task5b-bus-interrupted.log`,
and the isolated full rerun passed. All started runs were joined before completion.
Local commit only, no push, no live instance or credential-file 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.
2026-11-04T17:57:52Z | Codex implementation sub-agent, t3/lifecycle-task-6b-r2 | #797 / Gitea #1839 Task 6b rebase before/after | Before: local 347791b55 on the superseded Task 4 cherry-pick e41ec8159; origin/refactor is b5a58b48b4c254f166135c24e22bd63ce4b5e0b9 with Task 4 c098fa022, #772 793f3b157, Task 3 and Task 5b. After: replay only 347791b55 on the explicitly requested new branch t3/lifecycle-task-6b-r2; e41ec8159 is dropped. Shipped Task 3 migration 3q-3r and all seventeen existing migration files are byte-identical to the base. New 3r-3s rebuild preserves all fifteen shipped event kinds (including question.asked, question.answered, attach.selected and every Task 4 PRD/mode kind) and adds advanced, for sixteen; schema.sql matches, SCHEMA_VERSION=3s, SCHEMA_SHA256=fde7a46505a964d0e3802aaee96a1e2d9fb8c7163fbafc2f86aca914ecc50f03. The migration test carries rows of every prior kind, preserves a sqlite_sequence high-water mark, index and every request/effect guard, then appends advanced at the next sequence. Current-version, offline-chain and synthetic future-step test pins updated through 3s (future fixture 3t). Conflicts resolved by keeping both behavior sets: broker retains Console owner/verb guards, request question/answer/attach/cancel handlers, Task 5b builtin binding and scoped requirements/git admission, plus 6b advance and continuation services; lifecycle exports the union of Task 3 and 6b operations; runtime retains Task 5b builtin binding and #772 pending-product readiness while adding 6b recovery; harness keeps Task 5b project/snapshot/op schemas and catalog prose while adding 6b tools, with duplicate schema properties removed; six role duty documents merged cleanly and retained; append-only logs keep both sides. Integration corrections: unanswered owner questions suspend new continuation dispatch, answer evidence resumes it, and planning continuations carry the stable create op required by Task 5b. A real Task 5b adapter test creates/assigns the task with that op, advances it and keeps cross-project git refused. First full bus run 1554/1555 found a Task 3 global-zero-effects assertion; it now asserts attach preserves the exact preexisting continuation receipts. Another bus run and isolated #656 control reproduced the existing nice 16 versus 19 race: the installed Node scheduling rule changes startup priority. The owned FIFO child now sets priority after fixture construction immediately before the measured operation; all priority/nonblocking assertions remain intact, no daemon or live instance changed. FIFO 21/21 and final full bus 1555/1555 pass. First WebUI run 537/540, two failures and one skip: missing local esbuild 0.28.2 and one browser refresh timeout; npm ci --ignore-scripts installed only the locked WebUI dependencies, the browser file passes 2/2 in isolation, and whole WebUI passes 540/540 with MOSAIC_REQUIRE_EDITOR_BUILD=1 and test-concurrency=4. An accidental root npm test invocation reported no script and ran no tests; corrected to the bus package. Final GREEN, zero skips/cancellations: new 45/45, bus 1555/1555, harness 228/228, business 307/307, launch 8/8, whole WebUI 540/540. Every test invocation used its own mktemp -d /mnt/storage/scratch/tmp/XXXXXX TMPDIR. Logs /mnt/storage/scratch/tmp/task6b-r2-new-final.log, task6b-r2-bus-final.log, task6b-r2-harness.log, task6b-r2-business.log, task6b-r2-launch.log, task6b-r2-webui-final.log; complete per-run counts in task6b-r2-runs.json and schema evidence in task6b-r2-schema-proof.json in the same directory. No unresolved conflict, role JSON authority change, shipped migration edit, credential/config mount, push, merge, deployment or live-instance action. Queue command again refused this linked worktree; canonical coordination is unchanged. Local rebase only, with no attribution trailers.
+2
View File
@@ -2008,3 +2008,5 @@ are never rewritten or removed; corrections are new entries.
2026-11-04 | Claude Sonnet 5.5 implementation sub-agent, t3/console-request-intake-controls | #797 / Gitea #1839 lifecycle Task 3 | Intake Console controls and guided answers: 3p->3q migration, PM-only request.question, owner-only request.answer/attach/mode/cancel.console verbs, request-http.mjs routes with owner, same-origin and bounded-body checks, requests.js panel under the owner's chat message (mode choice through the decision confirmation, guided answers, project attach, mode change, cancel, reload-safe lost answer). Bus 1492/1492, webui 538/538, intake+bootstrap 70/70, zero skips; local commit only
2026-11-04T17:18Z | Claude Sonnet 5.5 implementation sub-agent, t3/console-request-intake-controls-r2 | #797 / Gitea #1839 lifecycle Task 3 rebase | Rebased 7fd0c3562 and c52218c65 onto c098fa022 (Task 4 core, #802). Migration renamed 3p-3q to 3q-3r, SCHEMA_VERSION 3r; Task 4's 3p-3q left byte-identical; the kind list is the union (Task 4's prd.submitted, prd.activated, prd.invalidated, mode.changed plus question.asked, question.answered, attach.selected). Task 3's own request.mode.console dropped for Task 4's modeConsole; webui maps request-mode-revision-conflict and reads Task 4's draft shape. Bus 1536/1536, webui 540/540, targeted bus 108/108 and webui 28/28, prd-activation 21/21, project prd 6/6, zero skips; not pushed.
- 2026-11-04 — Codex implementation sub-agent — Lifecycle Task 5b (#797/#1839): shared scoped catalogue admission, explicit validated Stack binding, TaskScope/provenance and Git project guards; based on 103168614 plus cherry-picked Task 4 1627e24d7. Final gates 35/107/132/1501 (+32 configuration/tools), zero skips; logs /mnt/storage/scratch/tmp/task5b-{newtests,regression,tasks,bus,tools}.log. No migration, local commit only, never pushed; no live instance touched.
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.
2026-11-04 | Codex implementation sub-agent, t3/lifecycle-task-6b-r2 | #797 / Gitea #1839 Task 6b rebase onto b5a58b48b | Dropped e41ec8159; merged Task 3 controls/questions, Task 5b scope/op admission and 6b continuations. New migration 3r-3s retains all fifteen shipped kinds plus advanced; all shipped migrations unchanged. Added question/op integration checks and fixed owned FIFO priority setup timing without weakening assertions. Final new 45/45, bus 1555/1555, harness 228/228, business 307/307, launch 8/8, whole WebUI 540/540, zero skips; per-run counts/log paths /mnt/storage/scratch/tmp/task6b-r2-runs.json. Both append-only logs retained; local rebase only, no attribution trailers or push/live changes; queue remains with canonical coordinator after linked-worktree refusal.
+1 -1
View File
@@ -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','question.asked','question.answered','attach.selected','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','question.asked','question.answered','attach.selected','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'),
+128 -7
View File
@@ -168,7 +168,7 @@ export const CONSOLE_VERBS = Object.freeze(['decision.preview.console', 'decisio
// What an owner's console cap may ask of the request lifecycle (a read, the intake controls, and Task 4's mode and draft).
const INTAKE_CONSOLE = Object.freeze(['get', 'answer', 'attach', 'cancel', 'mode', 'draft']);
const REQUEST_VERBS = Object.freeze({ 'request.open': 'open', 'request.get': 'get', 'request.projects': 'projects', 'request.propose': 'propose',
'prd.submit': 'submit', 'request.question': 'question', 'request.get.console': 'get', 'request.projects.console': 'projects',
'prd.submit': 'submit', 'request.advance': 'advance', 'request.question': 'question', 'request.get.console': 'get', 'request.projects.console': 'projects',
'request.answer.console': 'answer', 'request.attach.console': 'attach', 'request.cancel.console': 'cancel' });
// Row 275 (193b1b, #1780, row 193 design 4.3 and 9): the user verbs a viewer grant reaches. They
// read, so a live grant at any level covers them. thread.create, thread.open and thread.send need a
@@ -441,6 +441,7 @@ function checkBusiness(name, b) {
export class Broker {
#store;
#requests;
#requestRecoveryPending = false;
#businesses;
#sessions = new Map();
#runs = new Map();
@@ -610,6 +611,20 @@ export class Broker {
this.#requests = requestLifecycle({
dataRoot,
builtinBinding: builtinFor,
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) => {
@@ -652,6 +667,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');
@@ -672,7 +787,9 @@ export class Broker {
} else {
this.#agent(s);
if (INTAKE_CONSOLE.includes(operation) && operation !== 'get') fail('human-required');
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 });
@@ -832,6 +949,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
@@ -1855,6 +1973,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.
@@ -5277,7 +5397,7 @@ export class Broker {
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'], ['project_key']);
keys(args, ['project_key','requirement_snapshot','requirement'], ['project_key']);
return this.#requirementsFor(s, structuredClone(args))
.then(result => { this.#secretCheck(result); return result; })
.catch(e => { throw this.#refused(s, new BusError(e.code ?? 'requirement-snapshot-refused'), request); });
@@ -5288,7 +5408,7 @@ export class Broker {
if (!this.#requests) fail('request-not-configured');
const method = REQUEST_VERBS[request.verb];
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
@@ -5424,6 +5544,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;
});
@@ -7770,7 +7891,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(?,?,?,?,?,?,?,?,?,?,?)',
@@ -7784,9 +7905,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);
}
+46
View File
@@ -0,0 +1,46 @@
-- #797 / #1839 Task 6b: retain all fifteen shipped 3r event kinds and add the checked advanced event.
-- Preserve rows, seq, sqlite_sequence, the request index and every guard exactly as the prior rebuild.
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_3r;
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','question.asked','question.answered','attach.selected','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_3r 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_3r';
DROP TABLE request_events_3r;
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;
+6 -1
View File
@@ -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);
+150
View File
@@ -0,0 +1,150 @@
// 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.wait?.kind === 'question' && state.question?.answer === null) return [];
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,
question: state.question?.id ?? null, answer: state.question?.answer ?? 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, op: control.op, 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 });
}
+64 -8
View File
@@ -2,13 +2,15 @@
// PM's question and the owner's Console answer, existing-project pick and cancellation. The owner's mode change is Task 4's
// modeConsole, which moves the project's mode revision and every request attached to it.
// 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, positive, optionList, QUESTION_PHASES, PICKER_CODES } 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;
@@ -43,10 +45,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');
@@ -268,8 +282,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); }
@@ -304,6 +317,12 @@ 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);
});
}
// The Task 3 verbs share one shape: validate, authorize, replay a recorded op, then check revision and phase in
@@ -393,7 +412,29 @@ export function createRequestLifecycle({ journal, services }) {
});
});
}
return Object.freeze({ open, get, projects, propose, submit, question, answer, attach, cancel, 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, question, answer, attach, cancel, 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');
@@ -408,6 +449,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;
} });
}
+20 -2
View File
@@ -77,7 +77,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;
@@ -174,6 +174,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') {
// `decision` is the open selection decision the cancellation closes, or null when none is open. A project or
// mapping that already exists stays where it is; the journal never undoes an effect.
@@ -200,7 +207,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 => {
@@ -232,6 +247,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');
+15 -2
View File
@@ -284,7 +284,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. A managed product is part of it only with both a committed activate receipt in the
@@ -594,7 +594,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: services.builtinBinding });
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,
@@ -991,11 +991,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; },
@@ -1057,6 +1068,7 @@ export async function startBroker({
launches: bound,
readers: readCaps,
async close() {
clearInterval(requestTimer);
clearInterval(usageTimer);
await proxyUsage.close();
clearInterval(modelTimer);
@@ -1072,6 +1084,7 @@ export async function startBroker({
},
};
} catch (e) {
clearInterval(requestTimer);
clearInterval(usageTimer);
await proxyUsage?.close();
clearInterval(modelTimer);
+6 -2
View File
@@ -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 = '4dfc55b42a5967fb0ef2067040bd187db870596e2edda64e51f1a1025ada2092';
export const SCHEMA_VERSION = '3r';
export const SCHEMA_SHA256 = 'fde7a46505a964d0e3802aaee96a1e2d9fb8c7163fbafc2f86aca914ecc50f03';
export const SCHEMA_VERSION = '3s';
// 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';
@@ -103,6 +103,10 @@ const migrations = [{
from: '3q',
to: '3r',
sql: readFileSync(new URL('./migrations/3q-3r.sql', import.meta.url), 'utf8'),
}, {
// Task 6b: checked advancement after Task 3's complete fifteen-kind union.
from: '3r', to: '3s',
sql: readFileSync(new URL('./migrations/3r-3s.sql', import.meta.url), 'utf8'),
}];
const timedTables = [
'events',
@@ -92,6 +92,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) {
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'));
copyFileSync(here('../src/migrations/3r-3s.sql'), join(moduleRoot, 'src/migrations/3r-3s.sql'));
return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store;
}
@@ -99,7 +100,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), '3r');
assert.equal(version(path), '3s');
// 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]);
@@ -149,7 +150,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), '3r');
assert.equal(version(path), '3s');
assert.deepEqual(migrated3l(snapshot(path, before.columns).rows), before.rows);
});
@@ -161,14 +162,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 = '3r';", "export const SCHEMA_VERSION = '3h';")
.replace("export const SCHEMA_VERSION = '3s';", "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), '3r');
assert.equal(version(path), '3s');
assert.deepEqual(master(path), was);
assert.deepEqual(snapshot(path, after.columns).rows, after.rows);
});
+2 -2
View File
@@ -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 = '3r';", "export const SCHEMA_VERSION = '3c';")
.replace("export const SCHEMA_VERSION = '3s';", "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, '3r');
assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3s');
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 });
@@ -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, '3r');
assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3s');
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 = '3r';", "export const SCHEMA_VERSION = '3s';")
.replace('}];\nconst timedTables', `}, { from: '3r', to: '3s', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`);
.replace("export const SCHEMA_VERSION = '3s';", "export const SCHEMA_VERSION = '3t';")
.replace('}];\nconst timedTables', `}, { from: '3s', to: '3t', 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', '3q-3r.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', '3r-3s.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, '3s');
assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3t');
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 3r ready/);
assert.match(migrated.stdout, /bus schema 3s ready/);
new Store(root).close();
const usage = spawnSync(process.execPath, [command], { encoding: 'utf8' });
assert.equal(usage.status, 2);
+5 -4
View File
@@ -82,6 +82,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) {
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'));
copyFileSync(here('../src/migrations/3r-3s.sql'), join(moduleRoot, 'src/migrations/3r-3s.sql'));
return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store;
}
function fresh(t) {
@@ -103,7 +104,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), '3r');
assert.equal(version(path), '3s');
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)));
@@ -129,7 +130,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), '3r');
assert.equal(version(path), '3s');
});
test('the 3j engine refuses a migrated store, migrating or not, and leaves it as it was', async t => {
@@ -140,14 +141,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 = '3r';", "export const SCHEMA_VERSION = '3j';")
.replace("export const SCHEMA_VERSION = '3s';", "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), '3r');
assert.equal(version(path), '3s');
assert.deepEqual(master(path), was);
assert.deepEqual(rows(path), after);
});
+5 -4
View File
@@ -85,6 +85,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) {
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'));
copyFileSync(here('../src/migrations/3r-3s.sql'), join(moduleRoot, 'src/migrations/3r-3s.sql'));
return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store;
}
function fresh(t) {
@@ -108,7 +109,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), '3r');
assert.equal(version(path), '3s');
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)));
@@ -133,7 +134,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), '3r');
assert.equal(version(path), '3s');
});
test('the 3i engine refuses a migrated store, migrating or not, and leaves it as it was', async t => {
@@ -144,14 +145,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 = '3r';", "export const SCHEMA_VERSION = '3i';")
.replace("export const SCHEMA_VERSION = '3s';", "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), '3r');
assert.equal(version(path), '3s');
assert.deepEqual(master(path), was);
assert.deepEqual(rows(path), after);
});
@@ -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');
}
};
@@ -3,12 +3,14 @@ import fs from 'node:fs';
import { spawnSync } from 'node:child_process';
import { getPriority, setPriority } from 'node:os';
import { revocationFixture, outcome } from './model-revocation-fixture.mjs';
setPriority(0,19);
const cleanup=[];let answer;
try {
const f=await revocationFixture({after:close=>cleanup.push(close)},{onRoot:root=>process.send?.({kind:'owned-root',root})});
f.seed();fs.unlinkSync(f.temporary);const made=spawnSync('/usr/bin/mkfifo',['-m','600',f.temporary],{stdio:'ignore',timeout:5000,killSignal:'SIGTERM'});
if(made.status!==0)throw Error('owned fifo setup failed');
// Fixture setup can outlive a desktop scheduler's initial process classification.
// Set the measured priority at the controlled operation boundary, after setup.
setPriority(0,19);
await new Promise((resolve,reject)=>process.send({kind:'ready',nice:getPriority(0)},error=>error?reject(error):resolve()));
const began=Date.now(),result=await outcome(async()=>{await f.prepare();f.complete();});
answer={kind:'result',ok:result.ok,code:result.code??null,cachedRevoking:f.record()?.state==='revoking',finalPresent:f.exists(f.file),tmpPresent:f.exists(f.temporary),openModelFds:f.openModelFds(),elapsedMs:Date.now()-began,nice:getPriority(0)};f.valueFree();
+1 -1
View File
@@ -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, '3r');
assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3s');
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) });
+3 -3
View File
@@ -168,12 +168,12 @@ test('owner draft read shows exact bound PRD and requirements without a generic
for (const who of ['pm','human','reader']) assert.throws(() => f.call(who, 'request.draft.console', { request: f.request }), { code: 'human-required' });
});
test('3p to 3r migration preserves existing request rows and the append-only guards', async t => {
test('3p through 3s migration preserves existing request rows and the append-only guards', async t => {
const { readFileSync, chmodSync } = await import('node:fs');
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','question.asked','question.answered','attach.selected','mode.changed'", '');
.replace(",'prd.submitted','prd.activated','prd.invalidated','question.asked','question.answered','attach.selected','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 3r 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, '3r');
assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3s');
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/);
+8 -6
View File
@@ -182,6 +182,7 @@ test('an answer needs the outstanding question, a stale revision refuses, and a
test('attach records the owner pick from the authorized picker, then the PM proposes exactly that project for confirmation', async t => {
const x = await atPicker(t), request = x.source.request;
assert.equal(x.state.wait.code, 'request-existing-project');
const effectsBefore = x.store.all('SELECT * FROM request_effects ORDER BY seq');
const args = { request, revision: x.state.revision, project: 'old', op: 'attach-1' };
for (const project of ['library', 'unknown', 'factory', 'absent']) await assert.rejects(async () => x.call('console', 'request.attach.console', { ...args, project, op: 'attach-x' }), /request-project-refused/, project);
await assert.rejects(async () => x.call('console', 'request.attach.console', { ...args, revision: 1 }), /request-revision-conflict/);
@@ -192,8 +193,9 @@ test('attach records the owner pick from the authorized picker, then the PM prop
const ev = x.store.get("SELECT * FROM events WHERE kind='human.input' AND json_extract(body,'$.kind')='attach'");
assert.deepEqual(JSON.parse(ev.body), { kind: 'attach', request, project: 'old', op: 'attach-1', via: 'console', subject: SUBJECT, message: ev.subject });
assert.equal(x.store.get('SELECT to_role,from_run FROM messages WHERE id=?', ev.subject).to_role, 'pm');
// Nothing is attached yet: the pick stages no mapping, no effect, no decision.
assert.equal(n(x, 'SELECT count(*) n FROM request_effects'), 0); assert.equal(n(x, 'SELECT count(*) n FROM decisions'), 1);
// The pick stages no mapping, effect or decision; keep the existing continuation receipts unchanged.
assert.deepEqual(x.store.all('SELECT * FROM request_effects ORDER BY seq'), effectsBefore);
assert.equal(n(x, 'SELECT count(*) n FROM decisions'), 1);
await assert.rejects(async () => x.call('pm', 'request.propose', { ...x.existing(picked, 'existing-0'), project: 'library', source_project: 'library', name: 'Library' }), /request-attachment-mismatch|request-project-refused/);
const candidate = await x.call('pm', 'request.propose', x.existing(picked, 'existing-1'));
assert.equal(candidate.selection.bindings[0].plan.existing, true); assert.equal(candidate.project, null);
@@ -298,14 +300,14 @@ test('a cancel is refused when the owner-to-PM edge closed, with no withdrawn de
assert.equal((await controlUnderRefusedEdge(x, 'request.cancel.console', { request: x.source.request, revision: asked.revision, op: 'cancel-2' })).phase, 'cancelled');
});
test('3q to 3r widens the kind list to the union, keeps every 3q kind, row, seq and guard, and the chain matches the fresh digest', t => {
assert.equal(SCHEMA_VERSION, '3r');
test('3q through 3s widens the kind list to the union, keeps every 3q kind, row, seq and guard, and the chain matches the fresh digest', t => {
assert.equal(SCHEMA_VERSION, '3s');
const root = mkdtempSync(join(tmpdir(), 'request-migration-3r-'));
t.after(() => rmSync(root, { recursive: true, force: true })); mkdirSync(join(root, 'bus'), { mode: 0o700 });
const path = join(root, 'bus/bus.sqlite'), AT = '2026-11-03T00:00:00.000Z';
const current = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8');
// 3p-3q shipped the PRD kinds and mode.changed; 3q-3r adds the three Task 3 kinds after them.
const ddl3q = current.replace("'prd.invalidated','question.asked','question.answered','attach.selected','mode.changed','cancelled'", "'prd.invalidated','mode.changed','cancelled'");
const ddl3q = current.replace(",'advanced'", '').replace("'prd.invalidated','question.asked','question.answered','attach.selected','mode.changed','cancelled'", "'prd.invalidated','mode.changed','cancelled'");
assert.notEqual(ddl3q, current, 'the fixture is the narrower schema');
const migration = readFileSync(new URL('../src/migrations/3q-3r.sql', import.meta.url), 'utf8');
assert.ok(migration.includes("'prd.submitted','prd.activated','prd.invalidated','question.asked','question.answered','attach.selected','mode.changed','cancelled'"), 'the step carries every 3q kind plus its own');
@@ -322,7 +324,7 @@ test('3q to 3r widens the kind list to the union, keeps every 3q kind, row, seq
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 3r ready\n');
assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3s 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, including the 3q kinds');
const fresh = new DatabaseSync(':memory:'); fresh.exec(current);
@@ -0,0 +1,290 @@
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('3r to 3s 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', '3r');
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' }));
const oldKinds = ['opened','selection.proposed','selection.staged','selection.confirmed','setup.wait','bootstrap.stale','bootstrap.completed',
'prd.submitted','prd.activated','prd.invalidated','question.asked','question.answered','attach.selected','mode.changed','cancelled'];
for (const [n, kind] of oldKinds.entries())
db.prepare("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','r',?,?,?,'{}')").run(n + 1, kind, at);
db.prepare("UPDATE sqlite_sequence SET seq=41 WHERE name='request_events'").run();
const before = db.prepare('SELECT * FROM request_events ORDER BY seq').all();
const guards = db.prepare("SELECT name,sql FROM sqlite_master WHERE type='trigger' AND tbl_name IN ('request_events','request_effects') ORDER BY name").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 ORDER BY seq'), before);
assert.deepEqual(store.all("SELECT name,sql FROM sqlite_master WHERE type='trigger' AND tbl_name IN ('request_events','request_effects') ORDER BY name"), guards);
assert.equal(store.get("SELECT name FROM sqlite_master WHERE type='index' AND name='request_events_request'").name, 'request_events_request');
assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3s');
store.run("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','r',16,'advanced',?,'{}')", at);
assert.equal(store.get("SELECT max(seq) n FROM request_events").n, 42);
assert.throws(() => store.run("UPDATE request_events SET kind='opened'"), /append-only/);
assert.throws(() => store.run("INSERT OR REPLACE INTO request_events(seq,business,request,revision,kind,at,body) VALUES(1,'demo','r',1,'opened',?,'{}')", at), /append-only/);
assert.throws(() => store.run("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','r',20,'advanced',?,'{}')", at), /request-revision-conflict/);
assert.throws(() => store.run("INSERT INTO request_events(business,request,revision,kind,at,body) VALUES('demo','absent',1,'opened',?,'{}')", at), /request-origin-refused/);
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, []);
});
test('Task 3 question waits survive continuation recovery and resume after the owner answers', async t => {
const f = await requestFixture(t); await f.bus.requestLifecycle.recover();
const before = intents(f).length, state = f.journal.readRequest('demo', f.request);
const asked = await f.call('pm', 'request.question', { request: f.request, revision: state.revision, op: 'question1', text: 'Which timer duration?' });
await f.bus.requestLifecycle.recover();
assert.equal(intents(f).length, before, 'an owner question does not dispatch the next product operation');
const answered = await f.call('console', 'request.answer.console', { request: f.request, revision: asked.revision, question: asked.question.id, op: 'answer1', text: '25 minutes' });
await f.bus.requestLifecycle.recover();
const resumed = intents(f).find(e => e.result.intent.revision === answered.revision);
assert.equal(resumed.result.intent.step, 'submit_prd');
assert.equal(resumed.result.intent.evidence.answer, answered.question.answer);
});
test('Task 5b real scoped task creation and assignment feed the checked advance without weakening git scope', async t => {
const { scopedFixture } = await import('../../tasks/tests/scoped-fixture.mjs');
const f = await scopedFixture(t), p = await f.product('timer');
await f.bus.requestLifecycle.recover();
const planning = f.call('pm', 'message.receive').map(m => m.body.split('\n')[1]).filter(Boolean).map(line => { try { return JSON.parse(line); } catch { return null; } }).find(i => i?.next?.tool === 'task_create');
assert.equal(typeof planning.next.args.op, 'string', 'the resumed product create carries Task 5b stable operation identity');
const task = await f.create({ ...p.args, op: planning.next.args.op });
await f.bus.broker.requestTask(f.caps.pm, { verb: 'task.assign', args: { task_ref: task.task_ref, role: 'coder' } }, f.adapter.handle);
const state = await f.call('pm', 'request.get', { request: p.request });
const next = await f.call('pm', 'request.advance', { request: p.request, revision: state.revision, op: 'scoped-advance' });
assert.equal(next.phase, 'implementation');
assert.deepEqual(next.tasks, [task.task_ref]);
assert.throws(() => f.bus.checkPush(f.caps.coder, task.task_ref, 'stack'), { code: 'task-project-refused' });
assert.doesNotThrow(() => f.bus.checkPush(f.caps.coder, task.task_ref, 'timer'));
});
+55
View File
@@ -0,0 +1,55 @@
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) };
}
+5 -5
View File
@@ -273,8 +273,8 @@ test('plan_preparation_rejects_unknown_keys_changed_intent_tracker_modes_and_sym
test('3n_to_3o_is_explicit_offline_preserves_rows_and_old_engine_refuses_new_schema', t => {
const root = mkdtempSync(join(tmpdir(), 'request-migration-'));
t.after(() => rmSync(root, { recursive: true, force: true })); mkdirSync(join(root, 'bus'), { mode: 0o700 });
// Tasks 2, 3 and 4 (3o→3p, 3p→3q, 3q→3r) widen only request_events.kind, so the 3n 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','question.asked','question.answered','attach.selected','mode.changed','cancelled'", "'setup.wait','cancelled'");
// Tasks 2, 3, 4 and 6b (3o→3p, 3p→3q, 3q→3r, 3r→3s) widen only request_events.kind, so the 3n 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','question.asked','question.answered','attach.selected','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 3r ready\n'); assert.equal(SCHEMA_VERSION, '3r');
assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3s ready\n'); assert.equal(SCHEMA_VERSION, '3s');
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','question.asked','question.answered','attach.selected','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','question.asked','question.answered','attach.selected','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 3r ready\n');
assert.equal(cli.status, 0, cli.stderr); assert.equal(cli.stdout, 'bus schema 3s 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);
+1 -1
View File
@@ -525,7 +525,7 @@ test('v3b prototype refusals, views and append-only mutations', (t) => {
// Row 206 (#1704) adds thread_shares and its 3 append-only triggers; thread-shares.test.mjs tests them.
// Row 271 (#1776) adds session_ops, message_cancels and 7 triggers; session-clear.test.mjs tests them.
// #797 / #1839 Task 1 adds five append-only lifecycle tables and eighteen guards.
// Task 4 (3p to 3q) and Task 3 (3q to 3r) widen request_events.kind in place; all 30 tables and 114 guards remain.
// Task 4 (3p to 3q), Task 3 (3q to 3r) and Task 6b (3r to 3s) widen request_events.kind in place; all 30 tables and 114 guards remain.
assert.equal(count('table') - 1, 30);
assert.equal(count('trigger'), 114);
assert.equal(count('view'), 5);
+3 -3
View File
@@ -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), '3r');
assert.equal(version(path), '3s');
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 = '3r';", "export const SCHEMA_VERSION = '3m';")
.replace("export const SCHEMA_VERSION = '3s';", "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), '3r');
assert.equal(version(path), '3s');
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.
+3 -3
View File
@@ -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), '3r');
assert.equal(version(path), '3s');
// 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 = '3r';", "export const SCHEMA_VERSION = '3k';")
.replace("export const SCHEMA_VERSION = '3s';", "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), '3r');
assert.equal(version(path), '3s');
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.
+1 -1
View File
@@ -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, '3r');
assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3s');
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'), []);
+3 -3
View File
@@ -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 = '3r';", "export const SCHEMA_VERSION = '3d';")
.replace("export const SCHEMA_VERSION = '3s';", "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, '3r');
assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3s');
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 3r ready', ''].join('\n'));
assert.equal(migrated.stdout, [...REPORT, 'bus schema 3s ready', ''].join('\n'));
});
+3 -3
View File
@@ -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), '3r');
assert.equal(version(path), '3s');
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 = '3r';", "export const SCHEMA_VERSION = '3l';")
.replace("export const SCHEMA_VERSION = '3s';", "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), '3r');
assert.equal(version(path), '3s');
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.
+1 -1
View File
@@ -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, '3r');
assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3s');
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');
+78 -4
View File
@@ -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 });
}
});
}
+182
View File
@@ -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','[email protected]','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();
});
+1
View File
@@ -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 })),
+12 -2
View File
@@ -120,9 +120,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")) {
@@ -137,6 +137,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) {
+1 -1
View File
@@ -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);
+27 -2
View File
@@ -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');
}
});
+1 -1
View File
@@ -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",
+7
View File
@@ -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,
+13
View File
@@ -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
+9
View File
@@ -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
+10
View File
@@ -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.
+15
View File
@@ -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 <id>` from the user's message,
then `intake_status` to check its phase, revision and wait. Intake needs no
+8
View File
@@ -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