mirror of
https://github.com/PR0M3TH3AN/TORCH.git
synced 2026-11-06 21:01:54 +00:00
runtime: refill continuation slots and escalate unchanged work
This commit is contained in:
1 parent
f4253e6f5d
commit
512b5b6e16
5 files changed
+179
-24
No files matched your search
@@ -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,32 @@ 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'));
|
||||
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 +92,47 @@ 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;
|
||||
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) {
|
||||
|
||||
Reference in new issue
Block a user