mirror of
https://github.com/PR0M3TH3AN/TORCH.git
synced 2026-10-27 07:51:52 +00:00
Merge commit '981bc6933fb1db74e1ace8134741de298d49cfea' into torch/provider-runtime
This commit is contained in:
commit
d2d9845858
9 files changed
+392
-35
No files matched your search
@@ -1,5 +1,75 @@
|
||||
# Test Integrity Notes
|
||||
|
||||
## 2026-10-03 — Rolling continuation and bounded no-progress escalation
|
||||
|
||||
```yaml
|
||||
test_integrity_note:
|
||||
change_type: new_tests
|
||||
scenarios:
|
||||
- id: SCN-runtime-continuation-refill
|
||||
given: "A long manager turn and a short specialist turn occupy two slots; a third identity waits"
|
||||
when: "Only the short turn completes"
|
||||
then: "The third starts before manager completion and the daily attempt cap still holds"
|
||||
- id: SCN-runtime-continuation-no-progress
|
||||
given: "An unchanged unfinished eligible assignment"
|
||||
when: "Two successful executor turns make no recorded task progress"
|
||||
then: "Exactly two specialist turns occur, one durable coordination request is sent, unchanged-work repeats stop, new revisions may resume and unmet dependencies remain ineligible"
|
||||
- id: SCN-runtime-continuation-capacity
|
||||
given: "An owner-started executor already occupies the configured automatic capacity"
|
||||
when: "The continuation controller runs"
|
||||
then: "No additional automatic turn starts and no attempt is charged"
|
||||
- id: SCN-runtime-continuation-fairness
|
||||
given: "One specialist keeps advancing revisions while another eligible identity has not started"
|
||||
when: "A slot becomes available"
|
||||
then: "The least-recently-dispatched peer starts before the returning specialist"
|
||||
observable_outcomes:
|
||||
- "Actual launch ordering, daily charges, coordination messages and explicit held-work reason"
|
||||
determinism_controls:
|
||||
- "Hermetic installed Git projects, manually resolved provider promises and task fixtures; no sleep, network or live model"
|
||||
anti_cheat_rationale:
|
||||
prevents:
|
||||
- "Replacing the batch barrier by raising concurrency"
|
||||
- "Claiming refill only after all turns finish"
|
||||
- "Spending unlimited turns on unchanged work or ignoring dependency gates"
|
||||
- "Ignoring physical executors started outside the controller"
|
||||
relaxation:
|
||||
did_relax_any_assertion: false
|
||||
if_true_explain_spec_basis: ""
|
||||
```
|
||||
|
||||
## 2026-10-03 — Process-verified runtime presence
|
||||
|
||||
```yaml
|
||||
test_integrity_note:
|
||||
change_type: new_tests
|
||||
scenarios:
|
||||
- id: SCN-runtime-live-presence
|
||||
given: "A Linux executor holds a real identity guard while durable state is idle"
|
||||
when: "The observation API reads starting, working, approval waiting and terminal phases"
|
||||
then: "Live phases display accurately, semantic waits persist, terminal guards vanish and durable reports are not fabricated"
|
||||
- id: SCN-runtime-live-presence-refusal
|
||||
given: "A guarded identity with injected OS observations"
|
||||
when: "PID is replaced, dead, foreign-user, unavailable or project binding is wrong"
|
||||
then: "It never reports active execution"
|
||||
- id: SCN-runtime-live-presence-legacy
|
||||
given: "A pre-observation guard without a process-start token"
|
||||
when: "The exact TORCH up identity and canonical cwd match or disagree"
|
||||
then: "Only a verified matching live executor displays working"
|
||||
observable_outcomes:
|
||||
- "Observation API state, reportedState, executor source and unchanged durable identity"
|
||||
determinism_controls:
|
||||
- "Hermetic installed repository and injected PID observations for refusal cases"
|
||||
- "Supplemental Linux /proc integration observes the test process itself, with no provider or timing sleeps"
|
||||
anti_cheat_rationale:
|
||||
prevents:
|
||||
- "Painting all idle badges working without process evidence"
|
||||
- "Treating a crashed guard or reused PID as an active agent"
|
||||
- "Hiding approval waits or fabricating heartbeat/task progress"
|
||||
relaxation:
|
||||
did_relax_any_assertion: false
|
||||
if_true_explain_spec_basis: ""
|
||||
```
|
||||
|
||||
## 2026-09-30 — Bounded routine coordination
|
||||
|
||||
```yaml
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
# Live runtime presence instead of stale idle badges
|
||||
|
||||
The owner requires active executor turns to be distinguishable from old self-reported idle/offline state. Display is observational, not a new task-state machine or a substitute for authorization.
|
||||
|
||||
Each new physical turn guard records a starting phase and Linux process-start token, then enters working immediately before invoking the authorized executor. The read-only observation API validates the exact project/identity guard against a live same-user process with the matching start token. Existing legacy guards additionally require the exact TORCH up identity and canonical working directory. Dead, foreign-user, replaced or unobservable processes are not called active; retained unverified guards display unknown instead of false idle/offline.
|
||||
|
||||
An active verified guard overrides idle/offline display with starting/working. Original `reportedState`, `reportedSummary`, task and heartbeat are retained. Waiting/blocked reports remain visible even during a live turn. Guard removal naturally restores the last reported state after completion. No report, task, authority, launch policy, budget or receipt is changed by observation.
|
||||
|
||||
On unsupported OS process-observation surfaces, guarded execution remains unknown, not optimistically working. Other platform qualification remains open. Native harnesses with no resident TORCH executor may continue to rely on durable reports; this change does not claim to discover every external runtime process.
|
||||
|
||||
The live console pilot must load the source candidate to expose this observation. Updating its owned server does not upgrade the installed distribution or restart running agent turns. Broader startup/launch reservation work remains tracked by `TASK-pilot-starting-presence`.
|
||||
|
||||
Test integrity scenarios are recorded in docs/TEST_INTEGRITY.md. No existing expectations or test timeouts are relaxed.
|
||||
@@ -0,0 +1,17 @@
|
||||
# Rolling continuation without a whole-batch barrier
|
||||
|
||||
Owner-requested fix for `TASK-d95e72d0-816d-4fc2-acaf-b73642fd4881`.
|
||||
|
||||
The serialized controller reserves and charges each launch before dispatch. It waits for the first pending turn to complete, records its outcome, re-reads current policy/backlog/messages and immediately fills any available slot. A long manager turn no longer prevents other eligible identities from running. The controller lock still excludes competing automatic controllers; pending launches and all observed physical turn guards count toward automatic capacity. Owner-requested manual batches are separate authorization and are not silently included in the daily automatic attempt ledger.
|
||||
|
||||
The same approved roster, per-identity mutex, pause, detached state and daily cap remain authoritative. A pause prevents additional launches and never kills an already-started turn. Unknown or failed executor outcomes are not silently retried. Retained crashed guards/controller locks remain fail-closed and require supported recovery.
|
||||
|
||||
Successful turns may continue unchanged assigned/in-progress work if recorded dependencies are complete. After two consecutive successful turns leave the same task IDs/revisions unchanged, further unchanged-work retries are held and one durable request asks the Manager to resolve the blocker. New task revisions or fresh messages can justify another bounded turn. This is a conservative no-progress signal, not proof that no files changed or that a task is done; the Manager must inspect evidence. Waiting, blocked, unassigned, completed or unmet-dependency work is not automatically converted into implementation work. Existing one-assignment rules and self-claim policy are not changed.
|
||||
|
||||
Resume instructions now request eligible implementation/verification/coordination rather than just a status report. They do not grant additional ownership or approval authority. All output completion remains exact-evidence gated.
|
||||
|
||||
Status exposes occupied guard identities and no-progress holds. Installed release and reboot persistence remain unqualified. Existing native pilot service executes this source candidate on its next activation; the daily limit remains 12 automatic attempts, not a token-cost governor.
|
||||
|
||||
Eligible identities are ordered least-recently-dispatched within the current UTC day, with the Manager first only on ties. A repeatedly revised specialist task must not consume every freed slot while an eligible peer has never started.
|
||||
|
||||
Deterministic acceptance scenarios and anti-cheat notes are in docs/TEST_INTEGRITY.md. Independent live qualification remains necessary; do not close broader event-driven or persistent-operations umbrellas based on this source fix alone.
|
||||
+1
-1
@@ -1087,7 +1087,7 @@ function render(snapshot) {
|
||||
return `<article class="agent-row">
|
||||
<span class="presence ${escapeHtml(agent.state)}" aria-label="${live ? 'Active' : 'Offline'}"></span>
|
||||
<div class="agent-copy"><strong>${escapeHtml(agent.title ?? agent.areaId)}</strong><span>${escapeHtml(agent.summary ?? profile)}</span><small>${escapeHtml(agent.areaId)} · ${escapeHtml(profile)}</small></div>
|
||||
<div class="agent-state"><strong>${escapeHtml(agent.state ?? 'unknown')}</strong><small>${escapeHtml(ageLabel(agent.heartbeatAt))}</small></div>
|
||||
<div class="agent-state"><strong>${escapeHtml(agent.state ?? 'unknown')}</strong><small>${escapeHtml(agent.executor?.state === 'active' ? 'executor verified live' : ageLabel(agent.heartbeatAt))}</small></div>
|
||||
</article>`;
|
||||
}).join('') : empty('Install an approved Fleet to see persistent identities.'));
|
||||
|
||||
|
||||
@@ -16,6 +16,7 @@ import { assessBacklogHealth, backlogObservedCommitDistance } from '../backlog/s
|
||||
import { observeTaskActivity } from '../backlog/activity.mjs';
|
||||
import { ScheduleLauncherService } from '../schedules/launcher.mjs';
|
||||
import { hierarchyOrder } from '../runtime/hierarchy-order.mjs';
|
||||
import { observeRuntimeTurn } from '../runtime/turn-guard.mjs';
|
||||
|
||||
function json(path) {
|
||||
return JSON.parse(readFileSync(path, 'utf8'));
|
||||
@@ -417,11 +418,21 @@ export function observeProject({ repositoryRoot, env = process.env, now = () =>
|
||||
`).all().map((row) => [row.area_id, row]));
|
||||
result.agents = result.agents.map((agent) => {
|
||||
const row = live.get(agent.areaId);
|
||||
return row ? {
|
||||
const reported = row ? {
|
||||
...agent, state: row.state, runtime: row.runtime ?? agent.runtime,
|
||||
runtimeSessionId: row.runtime_session_id, summary: row.summary,
|
||||
task: row.task, heartbeatAt: row.heartbeat_at,
|
||||
} : agent;
|
||||
const executor = observeRuntimeTurn({ stateRoot, repositoryRoot: repository.root,
|
||||
projectId: manifest.projectId, areaId: agent.areaId, now });
|
||||
const staleIdle = ['idle', 'offline'].includes(reported.state);
|
||||
return { ...reported, reportedState: reported.state, reportedSummary: reported.summary, executor,
|
||||
...(executor.state === 'active' && staleIdle ? {
|
||||
state: executor.phase,
|
||||
summary: `Live runtime executor is ${executor.phase} (last reported state: ${reported.state}).`,
|
||||
} : executor.state === 'unknown' && staleIdle ? {
|
||||
state: 'unknown', summary: 'Runtime turn guard exists, but its executor could not be verified.',
|
||||
} : {}) };
|
||||
});
|
||||
}
|
||||
if (hasTable(database, 'schedule_wake_reservations')) {
|
||||
|
||||
@@ -1,17 +1,23 @@
|
||||
import { createHash, randomUUID } from 'node:crypto';
|
||||
import { existsSync, mkdirSync, readFileSync, renameSync, rmdirSync, writeFileSync } from 'node:fs';
|
||||
import { existsSync, mkdirSync, readFileSync, readdirSync, renameSync, rmdirSync, writeFileSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { TorchError } from '../kernel/errors.mjs';
|
||||
|
||||
function statePath(control) { return join(control.stateRoot, 'continuation.json'); }
|
||||
function occupiedAreas(control) {
|
||||
const path = join(control.stateRoot, 'sessions', 'turn-guards');
|
||||
return existsSync(path) ? readdirSync(path, { withFileTypes: true }).filter(d => d.isDirectory()).map(d => d.name) : [];
|
||||
}
|
||||
function taskKey(tasks) { return JSON.stringify(tasks.map(t => [t.id, t.revision]).sort((a, b) => a[0].localeCompare(b[0]))); }
|
||||
function read(control) {
|
||||
const policyPath = join(control.stateRoot, 'continuation-policy.json');
|
||||
const state = { enabled: false, maxTurnsPerDay: 12, areas: [], day: null, attempts: 0, handled: {},
|
||||
const state = { enabled: false, maxTurnsPerDay: 12, areas: [], day: null, attempts: 0, handled: {}, progress: {},
|
||||
...(existsSync(statePath(control)) ? JSON.parse(readFileSync(statePath(control), 'utf8')) : {}),
|
||||
...(existsSync(policyPath) ? JSON.parse(readFileSync(policyPath, 'utf8')) : {}) };
|
||||
if (typeof state.enabled !== 'boolean' || !Number.isInteger(state.maxTurnsPerDay)
|
||||
|| state.maxTurnsPerDay < 1 || state.maxTurnsPerDay > 48 || !Array.isArray(state.areas)
|
||||
|| !Number.isInteger(state.attempts) || state.attempts < 0 || !state.handled || typeof state.handled !== 'object') {
|
||||
|| !Number.isInteger(state.attempts) || state.attempts < 0 || !state.handled || typeof state.handled !== 'object'
|
||||
|| !state.progress || typeof state.progress !== 'object') {
|
||||
throw new TorchError('Continuation policy or attempt ledger is invalid; launch refused.', { code: 'CONTINUATION_POLICY_INVALID' });
|
||||
}
|
||||
return state;
|
||||
@@ -20,7 +26,7 @@ function save(control, state, policy = false) {
|
||||
const path = policy ? join(control.stateRoot, 'continuation-policy.json') : statePath(control);
|
||||
const temporary = `${path}.${randomUUID()}.tmp`;
|
||||
const value = policy ? { enabled: state.enabled, maxTurnsPerDay: state.maxTurnsPerDay, areas: state.areas }
|
||||
: { day: state.day, attempts: state.attempts, handled: state.handled };
|
||||
: { day: state.day, attempts: state.attempts, handled: state.handled, progress: state.progress };
|
||||
writeFileSync(temporary, JSON.stringify(value), { flag: 'wx', mode: 0o600 });
|
||||
renameSync(temporary, path);
|
||||
}
|
||||
@@ -48,22 +54,34 @@ export function planContinuation(control, backlog, { at = new Date(), limit = 3
|
||||
}
|
||||
const tasks = backlog.list();
|
||||
const candidates = [];
|
||||
const held = [];
|
||||
for (const areaId of state.areas) {
|
||||
const agent = control.getAgent(areaId);
|
||||
if (!['idle', 'offline', 'waiting'].includes(agent.state)
|
||||
|| existsSync(join(control.stateRoot, 'sessions', 'turn-guards', areaId))) continue;
|
||||
const messages = control.readMessages({ recipient: areaId, unacknowledgedOnly: true, limit: 1000 });
|
||||
const active = tasks.filter(t => t.owner === areaId && ['assigned', 'in_progress'].includes(t.state));
|
||||
const active = tasks.filter(t => t.owner === areaId && ['assigned', 'in_progress'].includes(t.state)
|
||||
&& (t.dependencies ?? []).every(id => tasks.find(t => t.id === id)?.state === 'completed'));
|
||||
const previous = state.handled[areaId] ?? {};
|
||||
const freshMessages = messages.filter(m => !previous.messageIds?.includes(m.id));
|
||||
const freshTasks = active.filter(t => previous.taskRevisions?.[t.id] !== t.revision);
|
||||
if (!freshMessages.length && !freshTasks.length) continue;
|
||||
const progress = state.progress[areaId];
|
||||
const continueWork = active.length > 0 && progress?.status === 0
|
||||
&& progress.taskKey === taskKey(active) && progress.noProgress < 2;
|
||||
if (!freshMessages.length && !freshTasks.length && !continueWork) {
|
||||
if (active.length && progress?.noProgress >= 2) held.push({ areaId, reason: 'no-progress-needs-coordination', taskIds: active.map(t => t.id) });
|
||||
continue;
|
||||
}
|
||||
const signal = { messageIds: messages.map(m => m.id), taskRevisions: Object.fromEntries(active.map(t => [t.id, t.revision])) };
|
||||
const fingerprint = createHash('sha256').update(JSON.stringify(signal)).digest('hex');
|
||||
candidates.push({ areaId, fingerprint, signal, messageCount: freshMessages.length, taskIds: freshTasks.map(t => t.id) });
|
||||
candidates.push({ areaId, fingerprint, signal, workKey: taskKey(active),
|
||||
messageCount: freshMessages.length, taskIds: active.map(t => t.id), continueWork });
|
||||
}
|
||||
candidates.sort((a, b) => Number(b.areaId === 'session-manager') - Number(a.areaId === 'session-manager'));
|
||||
const ordinal = area => state.handled[area]?.dispatchedDay === day ? state.handled[area].dispatchedOrdinal ?? 0 : 0;
|
||||
candidates.sort((a, b) => ordinal(a.areaId) - ordinal(b.areaId)
|
||||
|| Number(b.areaId === 'session-manager') - Number(a.areaId === 'session-manager'));
|
||||
return { enabled: true, day, attempts, remainingTurns: state.maxTurnsPerDay - attempts,
|
||||
occupiedAreas: occupiedAreas(control), held,
|
||||
candidates: candidates.slice(0, Math.min(limit, state.maxTurnsPerDay - attempts)), mutationPerformed: false };
|
||||
}
|
||||
|
||||
@@ -76,21 +94,48 @@ export async function runContinuation(control, backlog, { actorId, launch, at, c
|
||||
throw error;
|
||||
}
|
||||
try {
|
||||
const plan = planContinuation(control, backlog, { at: clock(), limit });
|
||||
const pending = [];
|
||||
for (const candidate of plan.candidates) {
|
||||
// Fresh plan and policy check before each launch; a concurrent pause wins.
|
||||
const current = planContinuation(control, backlog, { at: clock(), limit: 3 });
|
||||
if (!current.candidates.some(c => c.areaId === candidate.areaId && c.fingerprint === candidate.fingerprint)) continue;
|
||||
const pending = new Map();
|
||||
const launched = [];
|
||||
let reason = null;
|
||||
while (true) {
|
||||
// Refill immediately after EACH completion. Pending promises and physical
|
||||
// guards both count, including manual executors outside this controller.
|
||||
let current = planContinuation(control, backlog, { at: clock(), limit: 3 });
|
||||
let occupied = new Set([...occupiedAreas(control), ...pending.keys()]);
|
||||
while (occupied.size < limit && current.enabled && !current.reason) {
|
||||
const candidate = current.candidates.find(c => !occupied.has(c.areaId));
|
||||
if (!candidate) break;
|
||||
const state = read(control);
|
||||
if (state.day !== current.day) { state.day = current.day; state.attempts = 0; }
|
||||
state.attempts++;
|
||||
state.handled[candidate.areaId] = { ...candidate.signal,
|
||||
dispatchedDay: current.day, dispatchedOrdinal: state.attempts };
|
||||
save(control, state); // Reserve before launch; unknown attempts are never refunded.
|
||||
let result;
|
||||
try { result = launch(candidate.areaId); } catch { result = null; }
|
||||
pending.set(candidate.areaId, Promise.resolve(result).catch(() => null)
|
||||
.then(status => ({ areaId: candidate.areaId, status, candidate })));
|
||||
current = planContinuation(control, backlog, { at: clock(), limit: 3 });
|
||||
occupied = new Set([...occupiedAreas(control), ...pending.keys()]);
|
||||
}
|
||||
reason = current.reason ?? (occupied.size >= limit ? 'capacity-full' : null);
|
||||
if (!pending.size) break;
|
||||
const completed = await Promise.race(pending.values());
|
||||
pending.delete(completed.areaId);
|
||||
launched.push({ areaId: completed.areaId, status: completed.status });
|
||||
const state = read(control);
|
||||
if (state.day !== current.day) { state.day = current.day; state.attempts = 0; }
|
||||
state.attempts++;
|
||||
state.handled[candidate.areaId] = candidate.signal;
|
||||
save(control, state); // Charge/reserve before launch; interrupted attempts are never refunded/retried silently.
|
||||
let result;
|
||||
try { result = launch(candidate.areaId); } catch { result = null; }
|
||||
pending.push(Promise.resolve(result).catch(() => null).then(status => ({ areaId: candidate.areaId, status })));
|
||||
const active = backlog.list().filter(t => t.owner === completed.areaId && ['assigned', 'in_progress'].includes(t.state));
|
||||
const key = taskKey(active);
|
||||
const unchanged = active.length > 0 && key === completed.candidate.workKey;
|
||||
const prior = state.progress[completed.areaId];
|
||||
const noProgress = unchanged ? (prior?.taskKey === key ? prior.noProgress : 0) + 1 : 0;
|
||||
state.progress[completed.areaId] = { status: completed.status, taskKey: key, noProgress, completedAt: clock().toISOString() };
|
||||
save(control, state);
|
||||
if (completed.status === 0 && noProgress === 2 && typeof control.sendOwnerRequest === 'function') {
|
||||
control.sendOwnerRequest({ actorId, recipient: 'session-manager',
|
||||
body: `TORCH continuation requires coordination: ${completed.areaId} completed two turns without advancing assigned task state (${active.map(t => t.id).join(', ')}). Automatic unchanged-work retries are held. Inspect the blocker and route a concrete decision; do not mark work done without evidence.` });
|
||||
}
|
||||
}
|
||||
return { launched: await Promise.all(pending), reason: plan.reason ?? null };
|
||||
return { launched, reason };
|
||||
} finally { rmdirSync(lock); }
|
||||
}
|
||||
@@ -160,11 +160,11 @@ function startupMessage(areaId, fresh, instructionDigest) {
|
||||
if (areaId === 'session-manager') {
|
||||
return fresh
|
||||
? `Start the TORCH Fleet. Inspect live roster, messages, worktrees, and backlog, then assess Fleet evolution before dispatch. ${currentBrief}`
|
||||
: `Resume the TORCH Fleet from durable state. Reconcile live status and assess Fleet evolution before dispatch. ${currentBrief}`;
|
||||
: `Resume the TORCH Fleet from durable state. Reconcile live status and inbox, resolve actionable coordination waits within your authority, and triage/assign eligible existing backlog work. Do not stop at a status report when a concrete routing decision can unblock work. Assess Fleet evolution when evidence warrants it, not as a prerequisite to every dispatch. Preserve ownership, approvals and dependency gates. ${currentBrief}`;
|
||||
}
|
||||
return fresh
|
||||
? `Start this TORCH domain. Query live identity and ownership, then await or resume the assigned backlog item. ${currentBrief}`
|
||||
: `Resume this TORCH domain from durable state. Read the inbox, confirm ownership, and report current evidence. ${currentBrief}`;
|
||||
: `Resume this TORCH domain from durable state. Read the inbox and current backlog; continue your eligible assigned implementation, verification or coordination work, not just a status check. Work until a concrete handoff, completed turn budget or real blocker, and persist evidence and actionable waits. Never bypass dependencies, ownership or approval gates. If no eligible work exists, idle is valid. ${currentBrief}`;
|
||||
}
|
||||
|
||||
function atomicJson(path, value) {
|
||||
@@ -571,8 +571,9 @@ export function startFleet({ plan, controlPlane, executor, adapters } = {}) {
|
||||
const runtimes = adapterMap({ adapters });
|
||||
for (const action of plan.actions) {
|
||||
controlPlane.assertIdentity(action.areaId);
|
||||
withRuntimeTurnGuard(controlPlane, action.areaId, () => {
|
||||
withRuntimeTurnGuard(controlPlane, action.areaId, setPhase => {
|
||||
writeCombinedPrompt(action);
|
||||
setPhase('working');
|
||||
const result = executor({
|
||||
...action.launch, areaId: action.areaId, runtime: action.runtime,
|
||||
runtimeSessionId: action.runtimeSessionId,
|
||||
|
||||
@@ -1,8 +1,48 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { mkdirSync, readFileSync, rmdirSync, unlinkSync, writeFileSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { mkdirSync, readFileSync, readlinkSync, renameSync, rmdirSync, statSync, unlinkSync, writeFileSync } from 'node:fs';
|
||||
import { basename, join, resolve } from 'node:path';
|
||||
import { TorchError } from '../kernel/errors.mjs';
|
||||
|
||||
function nativeProcess(pid) {
|
||||
if (process.platform !== 'linux') throw new Error('Process observation unavailable');
|
||||
const directory = `/proc/${pid}`;
|
||||
const stat = readFileSync(join(directory, 'stat'), 'utf8');
|
||||
const fields = stat.slice(stat.lastIndexOf(')') + 2).split(' ');
|
||||
return { uid: statSync(directory).uid, state: fields[0], startTicks: fields[19],
|
||||
cwd: readlinkSync(join(directory, 'cwd')),
|
||||
argv: readFileSync(join(directory, 'cmdline'), 'utf8').split('\0').filter(Boolean) };
|
||||
}
|
||||
|
||||
// Observational only: never changes assignments, heartbeat, authority or launch eligibility.
|
||||
export function observeRuntimeTurn({ stateRoot, repositoryRoot, projectId, areaId,
|
||||
inspectProcess = nativeProcess, uid = process.getuid?.(), now = () => new Date() } = {}) {
|
||||
if (!/^[a-z0-9][a-z0-9-]*$/.test(areaId)) return { state: 'unknown', reason: 'invalid-identity' };
|
||||
let record;
|
||||
try { record = JSON.parse(readFileSync(join(stateRoot, 'sessions', 'turn-guards', areaId, 'owner.json'), 'utf8')); }
|
||||
catch (error) { return { state: error.code === 'ENOENT' ? 'inactive' : 'unknown', reason: 'guard-unavailable' }; }
|
||||
if (record.projectId !== projectId || record.areaId !== areaId || !Number.isSafeInteger(record.pid)
|
||||
|| record.pid < 1 || typeof record.nonce !== 'string' || !record.nonce) {
|
||||
return { state: 'unknown', reason: 'guard-invalid' };
|
||||
}
|
||||
try {
|
||||
const live = inspectProcess(record.pid);
|
||||
if (live.uid !== uid || ['Z', 'X'].includes(live.state)) return { state: 'unknown', reason: 'executor-not-live' };
|
||||
if (record.startTicks) {
|
||||
if (record.startTicks !== live.startTicks) return { state: 'unknown', reason: 'executor-replaced' };
|
||||
} else {
|
||||
// Existing pre-observation guards: require an exact identity-bound TORCH up process.
|
||||
const only = live.argv.indexOf('--only');
|
||||
if (resolve(live.cwd) !== resolve(repositoryRoot) || !live.argv.includes('up')
|
||||
|| only < 0 || live.argv[only + 1] !== areaId
|
||||
|| !live.argv.some(arg => ['torch.mjs', 'torch'].includes(basename(arg)))) {
|
||||
return { state: 'unknown', reason: 'legacy-executor-unverified' };
|
||||
}
|
||||
}
|
||||
return { state: 'active', phase: record.phase === 'starting' ? 'starting' : 'working',
|
||||
observedAt: now().toISOString(), source: 'live-process-and-turn-guard' };
|
||||
} catch { return { state: 'unknown', reason: 'executor-unobservable' }; }
|
||||
}
|
||||
|
||||
// One physical executor per identity, including the interval after an agent
|
||||
// reports idle but before its harness actually exits. Crashed guards never expire.
|
||||
export function withRuntimeTurnGuard(controlPlane, areaId, run) {
|
||||
@@ -19,8 +59,19 @@ export function withRuntimeTurnGuard(controlPlane, areaId, run) {
|
||||
}
|
||||
const path = join(directory, 'owner.json');
|
||||
const nonce = randomUUID();
|
||||
writeFileSync(path, JSON.stringify({ nonce, pid: process.pid, areaId, projectId: controlPlane.projectId }), { flag: 'wx', mode: 0o600 });
|
||||
try { return run(); } finally {
|
||||
let startTicks;
|
||||
try { startTicks = nativeProcess(process.pid).startTicks; } catch { /* Unavailable hosts remain observationally unknown. */ }
|
||||
const record = { nonce, pid: process.pid, areaId, projectId: controlPlane.projectId, startTicks, phase: 'starting' };
|
||||
writeFileSync(path, JSON.stringify(record), { flag: 'wx', mode: 0o600 });
|
||||
const setPhase = phase => {
|
||||
if (!['starting', 'working'].includes(phase)) throw new TorchError('Invalid runtime phase.', { code: 'INVALID_RUNTIME_INPUT' });
|
||||
const current = JSON.parse(readFileSync(path, 'utf8'));
|
||||
if (current.nonce !== nonce || current.pid !== process.pid) throw new TorchError('Runtime guard changed.', { code: 'RUNTIME_TURN_ACTIVE' });
|
||||
const temporary = `${path}.${nonce}.tmp`;
|
||||
writeFileSync(temporary, JSON.stringify({ ...record, phase }), { flag: 'wx', mode: 0o600 });
|
||||
renameSync(temporary, path);
|
||||
};
|
||||
try { return run(setPhase); } finally {
|
||||
// Delete only the exact guard created by this invocation; preserve foreign/changed guards.
|
||||
const record = JSON.parse(readFileSync(path, 'utf8'));
|
||||
if (record.nonce === nonce && record.pid === process.pid) {
|
||||
|
||||
@@ -10,10 +10,11 @@ import { proposeDomains } from '../../src/kernel/domains.mjs';
|
||||
import { installProject } from '../../src/kernel/install.mjs';
|
||||
import { openControlPlane } from '../../src/control-plane/service.mjs';
|
||||
import { issueStoppedExecutorEvidence } from '../../src/runtime/stopped-executor-evidence.mjs';
|
||||
import { withRuntimeTurnGuard } from '../../src/runtime/turn-guard.mjs';
|
||||
import { observeRuntimeTurn, withRuntimeTurnGuard } from '../../src/runtime/turn-guard.mjs';
|
||||
import { observeProject } from '../../src/observability/snapshot.mjs';
|
||||
import { configureContinuation, planContinuation, runContinuation } from '../../src/runtime/continuation.mjs';
|
||||
|
||||
function fixture() {
|
||||
function fixture({ recoveryWorktree = true, extraWorker = false } = {}) {
|
||||
const root = mkdtempSync(join(tmpdir(), 'torch-legacy-recovery-'));
|
||||
for (const args of [['init', '-b', 'main'], ['config', 'user.email', '[email protected]'],
|
||||
['config', 'user.name', 'Recovery Fixture']]) execFileSync('git', args, { cwd: root, stdio: 'pipe' });
|
||||
@@ -22,13 +23,14 @@ function fixture() {
|
||||
execFileSync('git', ['commit', '-m', 'fixture'], { cwd: root, stdio: 'pipe' });
|
||||
const repository = inspectRepository(root);
|
||||
const proposal = proposeDomains({ repository, analysis: analyzeRepository(repository) });
|
||||
if (extraWorker) proposal.domains.push({ ...proposal.domains[0], id: 'extra-worker', title: 'Extra worker', owned_paths: ['extra/**'] });
|
||||
proposal.review = { status: 'approved', reviewedAt: '2026-10-03T00:00:00Z', reviewedBy: 'owner', notes: [] };
|
||||
const env = { ...process.env, XDG_DATA_HOME: join(root, '.data') };
|
||||
installProject({ repository, proposal, env, projectId: 'recovery-fixture' });
|
||||
const actualPath = join(root, '.torch', 'install-manifest.json');
|
||||
const manifest = JSON.parse(readFileSync(actualPath, 'utf8'));
|
||||
const areaId = proposal.domains[0].id;
|
||||
manifest.external.push({ type: 'worktree', area: areaId, path: join(root, 'worker') });
|
||||
if (recoveryWorktree) manifest.external.push({ type: 'worktree', area: areaId, path: join(root, 'worker') });
|
||||
writeFileSync(actualPath, JSON.stringify(manifest));
|
||||
let time = Date.parse('2026-10-03T00:00:00Z');
|
||||
const control = openControlPlane({ repositoryRoot: root, env, clock: () => new Date(time) });
|
||||
@@ -44,7 +46,7 @@ function fixture() {
|
||||
invocationId: 'a'.repeat(32), readUnit: () => ({ ...properties }), readProcesses: () => processes, now: () => time });
|
||||
const apply = issued => control.recoverOwnerStoppedUnknownIdentity({ actorId: 'owner', areaId,
|
||||
expectedIdentity: issued.plan.expectedIdentity, evidence: issued.evidence });
|
||||
return { control, identity, areaId, properties, issue, apply,
|
||||
return { root, env, control, identity, areaId, properties, issue, apply,
|
||||
advance: ms => { time += ms; }, processes: value => { processes = value; } };
|
||||
}
|
||||
|
||||
@@ -122,6 +124,75 @@ test('SCN-runtime-turn-guard: overlapping executors refuse even if an agent repo
|
||||
} finally { f.control.close(); }
|
||||
});
|
||||
|
||||
test('SCN-runtime-live-presence: live guard phases override stale idle display without changing durable reports or waits', { skip: process.platform !== 'linux' }, () => {
|
||||
const f = fixture({ recoveryWorktree: false });
|
||||
try {
|
||||
f.control.reportStatus({ areaId: f.areaId, state: 'idle', summary: 'Previous turn completed' });
|
||||
const observe = () => observeProject({ repositoryRoot: f.root, env: f.env }).agents.find(a => a.areaId === f.areaId);
|
||||
withRuntimeTurnGuard(f.control, f.areaId, phase => {
|
||||
let agent = observe();
|
||||
assert.equal(agent.state, 'starting');
|
||||
assert.equal(agent.reportedState, 'idle');
|
||||
assert.equal(agent.executor.source, 'live-process-and-turn-guard');
|
||||
phase('working');
|
||||
agent = observe();
|
||||
assert.equal(agent.state, 'working');
|
||||
assert.equal(f.control.getAgent(f.areaId).state, 'idle', 'Observation is not an identity mutation');
|
||||
f.control.reportStatus({ areaId: f.areaId, state: 'waiting', summary: 'Waiting for named approval' });
|
||||
assert.equal(observe().state, 'waiting', 'Live execution must not conceal semantic approval waits');
|
||||
});
|
||||
assert.equal(observe().executor.state, 'inactive');
|
||||
assert.equal(observe().state, 'waiting');
|
||||
f.control.reportStatus({ areaId: f.areaId, state: 'idle', summary: 'Turn completed' });
|
||||
assert.equal(observe().state, 'idle');
|
||||
} finally { f.control.close(); }
|
||||
});
|
||||
|
||||
test('SCN-runtime-live-presence-refusal: dead, replaced, foreign-user and malformed guards never claim active execution', () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
withRuntimeTurnGuard(f.control, f.areaId, () => {
|
||||
const path = join(f.control.stateRoot, 'sessions', 'turn-guards', f.areaId, 'owner.json');
|
||||
const original = JSON.parse(readFileSync(path, 'utf8'));
|
||||
writeFileSync(path, JSON.stringify({ ...original, startTicks: 'fixture-start' }));
|
||||
const read = live => observeRuntimeTurn({ stateRoot: f.control.stateRoot, repositoryRoot: f.root,
|
||||
projectId: f.control.projectId, areaId: f.areaId, uid: 1000, inspectProcess: () => live });
|
||||
const live = { uid: 1000, startTicks: 'fixture-start', state: 'S' };
|
||||
assert.equal(read(live).state, 'active');
|
||||
for (const mismatch of [{ ...live, uid: 2000 }, { ...live, startTicks: 'reused-pid' }, { ...live, state: 'Z' }]) {
|
||||
assert.equal(read(mismatch).state, 'unknown');
|
||||
}
|
||||
const absent = observeRuntimeTurn({ stateRoot: f.control.stateRoot, repositoryRoot: f.root,
|
||||
projectId: f.control.projectId, areaId: f.areaId, inspectProcess: () => { throw new Error('Process gone'); } });
|
||||
assert.equal(absent.state, 'unknown');
|
||||
writeFileSync(path, JSON.stringify({ ...original, projectId: 'foreign' }));
|
||||
assert.equal(read(live).reason, 'guard-invalid');
|
||||
writeFileSync(path, JSON.stringify(original));
|
||||
});
|
||||
} finally { f.control.close(); }
|
||||
});
|
||||
|
||||
test('SCN-runtime-live-presence-legacy: pre-observation guards require exact TORCH identity and project cwd', () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
withRuntimeTurnGuard(f.control, f.areaId, () => {
|
||||
const path = join(f.control.stateRoot, 'sessions', 'turn-guards', f.areaId, 'owner.json');
|
||||
const original = JSON.parse(readFileSync(path, 'utf8'));
|
||||
const legacy = { ...original };
|
||||
delete legacy.startTicks;
|
||||
delete legacy.phase;
|
||||
writeFileSync(path, JSON.stringify(legacy));
|
||||
const inspect = live => observeRuntimeTurn({ stateRoot: f.control.stateRoot, repositoryRoot: f.root,
|
||||
projectId: f.control.projectId, areaId: f.areaId, uid: 1000, inspectProcess: () => live });
|
||||
const live = { uid: 1000, state: 'S', cwd: f.root, argv: ['node', '/runtime/bin/torch.mjs', 'up', '--only', f.areaId] };
|
||||
assert.equal(inspect(live).phase, 'working');
|
||||
assert.equal(inspect({ ...live, cwd: '/foreign' }).state, 'unknown');
|
||||
assert.equal(inspect({ ...live, argv: ['node', 'torch.mjs', 'up', '--only', 'someone-else'] }).state, 'unknown');
|
||||
writeFileSync(path, JSON.stringify(original));
|
||||
});
|
||||
} finally { f.control.close(); }
|
||||
});
|
||||
|
||||
test('SCN-runtime-continuation: explicit enable, event deduplication, daily cap and pause govern authorized turns', async () => {
|
||||
const f = fixture();
|
||||
const backlog = { list: () => [] };
|
||||
@@ -183,6 +254,84 @@ test('SCN-runtime-continuation-pause: a pause during one turn prevents another q
|
||||
} finally { f.control.close(); }
|
||||
});
|
||||
|
||||
test('SCN-runtime-continuation-refill: a short turn frees capacity while the manager remains running', async () => {
|
||||
const f = fixture({ extraWorker: true });
|
||||
try {
|
||||
f.control.reportStatus({ areaId: f.areaId, state: 'idle', summary: 'Idle' });
|
||||
for (const recipient of ['session-manager', f.areaId, 'extra-worker']) {
|
||||
f.control.sendOwnerRequest({ actorId: 'owner', recipient, body: 'Eligible handoff' });
|
||||
}
|
||||
configureContinuation(f.control, { actorId: 'owner', enabled: true, maxTurnsPerDay: 3 });
|
||||
const starts = [];
|
||||
const finishes = new Map();
|
||||
let thirdStarted;
|
||||
const third = new Promise(resolve => { thirdStarted = resolve; });
|
||||
const run = runContinuation(f.control, { list: () => [] }, { actorId: 'owner', limit: 2,
|
||||
launch: area => { starts.push(area); if (starts.length === 3) thirdStarted();
|
||||
return new Promise(resolve => finishes.set(area, resolve)); } });
|
||||
assert.deepEqual(starts, ['session-manager', f.areaId]);
|
||||
finishes.get(f.areaId)(0);
|
||||
await third;
|
||||
assert.deepEqual(starts, ['session-manager', f.areaId, 'extra-worker']);
|
||||
assert.equal(finishes.has('session-manager'), true, 'Manager completion was not required for refill');
|
||||
finishes.get('extra-worker')(0);
|
||||
finishes.get('session-manager')(0);
|
||||
assert.equal((await run).launched.length, 3);
|
||||
assert.equal(planContinuation(f.control, { list: () => [] }).reason, 'daily-turn-cap');
|
||||
} finally { f.control.close(); }
|
||||
});
|
||||
|
||||
test('SCN-runtime-continuation-no-progress: unfinished assignment continues once then escalates rather than busy-looping', async () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.control.reportStatus({ areaId: f.areaId, state: 'idle', summary: 'Idle' });
|
||||
configureContinuation(f.control, { actorId: 'owner', enabled: true, maxTurnsPerDay: 8 });
|
||||
const task = { id: 'TASK-continue', owner: f.areaId, state: 'assigned', revision: 1, dependencies: [] };
|
||||
let specialistTurns = 0;
|
||||
const result = await runContinuation(f.control, { list: () => [task] }, { actorId: 'owner', limit: 1,
|
||||
launch: area => { if (area === f.areaId) specialistTurns++; return 0; } });
|
||||
assert.equal(specialistTurns, 2);
|
||||
assert.equal(result.launched.filter(t => t.areaId === f.areaId).length, 2);
|
||||
const messages = f.control.readMessages({ recipient: 'session-manager', unacknowledgedOnly: true, limit: 1000 });
|
||||
assert.equal(messages.filter(m => m.body.includes('two turns without advancing')).length, 1);
|
||||
assert.equal(planContinuation(f.control, { list: () => [task] }).candidates.some(c => c.areaId === f.areaId), false);
|
||||
task.revision++;
|
||||
assert.equal(planContinuation(f.control, { list: () => [task] }).candidates.some(c => c.areaId === f.areaId), true);
|
||||
task.dependencies = ['TASK-unfinished'];
|
||||
assert.equal(planContinuation(f.control, { list: () => [task] }).candidates.some(c => c.areaId === f.areaId), false);
|
||||
} finally { f.control.close(); }
|
||||
});
|
||||
|
||||
test('SCN-runtime-continuation-capacity: a manually held executor counts against automatic fleet capacity', async () => {
|
||||
const f = fixture();
|
||||
try {
|
||||
f.control.reportStatus({ areaId: f.areaId, state: 'idle', summary: 'Idle' });
|
||||
f.control.sendOwnerRequest({ actorId: 'owner', recipient: f.areaId, body: 'Eligible handoff' });
|
||||
configureContinuation(f.control, { actorId: 'owner', enabled: true });
|
||||
let running;
|
||||
withRuntimeTurnGuard(f.control, 'session-manager', () => {
|
||||
running = runContinuation(f.control, { list: () => [] }, { actorId: 'owner', limit: 1,
|
||||
launch: () => { throw new Error('Capacity must not be exceeded'); } });
|
||||
});
|
||||
assert.deepEqual((await running).launched, []);
|
||||
assert.equal(planContinuation(f.control, { list: () => [] }).attempts, 0);
|
||||
} finally { f.control.close(); }
|
||||
});
|
||||
|
||||
test('SCN-runtime-continuation-fairness: a fresh task revision does not let one specialist starve an unstarted peer', async () => {
|
||||
const f = fixture({ extraWorker: true });
|
||||
try {
|
||||
f.control.reportStatus({ areaId: f.areaId, state: 'idle', summary: 'Idle' });
|
||||
configureContinuation(f.control, { actorId: 'owner', enabled: true, maxTurnsPerDay: 3 });
|
||||
const task = { id: 'TASK-changing', owner: f.areaId, state: 'assigned', revision: 1, dependencies: [] };
|
||||
f.control.sendOwnerRequest({ actorId: 'owner', recipient: 'extra-worker', body: 'Peer work' });
|
||||
const starts = [];
|
||||
await runContinuation(f.control, { list: () => [task] }, { actorId: 'owner', limit: 1,
|
||||
launch: area => { starts.push(area); if (area === f.areaId) task.revision++; return 0; } });
|
||||
assert.deepEqual(starts.slice(0, 2), [f.areaId, 'extra-worker']);
|
||||
} finally { f.control.close(); }
|
||||
});
|
||||
|
||||
test('SCN-legacy-runtime-recovery-project: evidence cannot be applied to a different checkout with cloned project/session fields', () => {
|
||||
const first = fixture();
|
||||
const second = fixture();
|
||||
|
||||
Reference in new issue
Block a user