mirror of
https://github.com/PR0M3TH3AN/TORCH.git
synced 2026-10-27 16:01:54 +00:00
345 lines
16 KiB
JavaScript
345 lines
16 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import { execFileSync } from 'node:child_process';
|
|
import { existsSync, mkdtempSync, readFileSync, writeFileSync } from 'node:fs';
|
|
import { tmpdir } from 'node:os';
|
|
import { join } from 'node:path';
|
|
import test from 'node:test';
|
|
import { openControlPlane } from '../../src/control-plane/service.mjs';
|
|
import { analyzeRepository } from '../../src/kernel/analyze.mjs';
|
|
import { proposeDomains } from '../../src/kernel/domains.mjs';
|
|
import { inspectRepository } from '../../src/kernel/git.mjs';
|
|
import { installProject } from '../../src/kernel/install.mjs';
|
|
|
|
function installedFixture() {
|
|
const root = mkdtempSync(join(tmpdir(), 'torch-control-plane-'));
|
|
execFileSync('git', ['init', '-b', 'main', root]);
|
|
execFileSync('git', ['-C', root, 'config', 'user.email', '[email protected]']);
|
|
execFileSync('git', ['-C', root, 'config', 'user.name', 'TORCH Test']);
|
|
writeFileSync(join(root, 'index.js'), 'export const ready = true;\n');
|
|
writeFileSync(join(root, 'index.test.js'), 'export const scenario = true;\n');
|
|
execFileSync('git', ['-C', root, 'add', '.']);
|
|
execFileSync('git', ['-C', root, 'commit', '-m', 'fixture']);
|
|
|
|
const repository = inspectRepository(root);
|
|
const proposal = proposeDomains({ repository, analysis: analyzeRepository(repository) });
|
|
proposal.review = {
|
|
status: 'approved', reviewedAt: '2026-09-27T00:00:00Z', reviewedBy: 'fixture-owner', notes: [],
|
|
};
|
|
const env = { ...process.env, XDG_DATA_HOME: join(root, '.data') };
|
|
installProject({ repository, proposal, env, projectId: 'control-plane-fixture' });
|
|
const areaIds = proposal.domains.map((domain) => domain.id);
|
|
return { root, env, worker: areaIds.find((id) => id !== 'qa') ?? areaIds[0], qa: areaIds.find((id) => id === 'qa') };
|
|
}
|
|
|
|
function deterministicOptions(fixture) {
|
|
let sequence = 0;
|
|
let seconds = 0;
|
|
return {
|
|
repositoryRoot: fixture.root,
|
|
env: fixture.env,
|
|
idFactory: () => `event-${++sequence}`,
|
|
clock: () => new Date(Date.UTC(2026, 8, 27, 12, 0, seconds++)),
|
|
};
|
|
}
|
|
|
|
function writeCapableSql(sql) {
|
|
return /(?:^\s*|[);]\s*)(?:INSERT|UPDATE|DELETE|REPLACE|CREATE|ALTER|DROP|VACUUM)\b/i.test(sql);
|
|
}
|
|
|
|
function quoteIdentifier(identifier) {
|
|
return `"${identifier.replaceAll('"', '""')}"`;
|
|
}
|
|
|
|
function durableObservationState(control, recipient) {
|
|
const tables = control.database.prepare(`
|
|
SELECT name FROM sqlite_master
|
|
WHERE type = 'table' AND name NOT LIKE 'sqlite_%'
|
|
ORDER BY name ASC
|
|
`).all();
|
|
const durableTables = Object.fromEntries(tables.map(({ name }) => [name,
|
|
control.database.prepare(`SELECT * FROM ${quoteIdentifier(name)} ORDER BY rowid ASC`).all(),
|
|
]));
|
|
const visibleMessages = control.database.prepare(`
|
|
SELECT m.*, a.acknowledged_at
|
|
FROM messages m
|
|
LEFT JOIN message_acks a ON a.message_id = m.id AND a.area_id = ?
|
|
WHERE m.recipient_id = ? OR m.recipient_id = 'all'
|
|
ORDER BY m.created_at ASC, m.id ASC
|
|
`).all(recipient, recipient);
|
|
return {
|
|
durableTables,
|
|
visibleMessages,
|
|
ownership: control.whoOwns({ path: 'index.js' }),
|
|
};
|
|
}
|
|
|
|
function observeConnectionWrites(control) {
|
|
const database = control.database;
|
|
const writeCalls = [];
|
|
const instrumentStatement = (statement, sql) => new Proxy(statement, {
|
|
get(target, property) {
|
|
const value = Reflect.get(target, property, target);
|
|
if (writeCapableSql(sql) && ['run', 'all', 'get', 'iterate'].includes(property)) {
|
|
return (...args) => {
|
|
writeCalls.push({ method: property, sql: sql.replace(/\s+/g, ' ').trim() });
|
|
return value.apply(target, args);
|
|
};
|
|
}
|
|
return typeof value === 'function' ? value.bind(target) : value;
|
|
},
|
|
});
|
|
control.database = new Proxy(database, {
|
|
get(target, property) {
|
|
const value = Reflect.get(target, property, target);
|
|
if (property === 'prepare') return (sql) => instrumentStatement(value.call(target, sql), sql);
|
|
if (property === 'exec') {
|
|
return (sql) => {
|
|
if (writeCapableSql(sql)) writeCalls.push({ method: 'exec', sql: sql.replace(/\s+/g, ' ').trim() });
|
|
return value.call(target, sql);
|
|
};
|
|
}
|
|
return typeof value === 'function' ? value.bind(target) : value;
|
|
},
|
|
});
|
|
return {
|
|
writeCalls,
|
|
restore() { control.database = database; },
|
|
};
|
|
}
|
|
|
|
function observeWithoutDurableWrite(control, fixture, input) {
|
|
const beforeState = durableObservationState(control, fixture.worker);
|
|
const beforeBytes = readFileSync(join(control.stateRoot, 'state.db'));
|
|
const observer = observeConnectionWrites(control);
|
|
let observation;
|
|
try {
|
|
observation = control.observeMessages(input);
|
|
} finally {
|
|
observer.restore();
|
|
}
|
|
assert.equal(observer.writeCalls.length, 0,
|
|
`observeMessages executed ${observer.writeCalls.length} DML write execution(s): ${JSON.stringify(observer.writeCalls)}`);
|
|
assert.deepEqual(durableObservationState(control, fixture.worker), beforeState,
|
|
'observing messages must preserve message visibility/order and every logical durable fixture record');
|
|
assert.deepEqual(readFileSync(join(control.stateRoot, 'state.db')), beforeBytes,
|
|
'observing messages must leave the main SQLite database bytes unchanged');
|
|
return observation;
|
|
}
|
|
|
|
test('SCN-durable-message: direct and group messages survive restart and require recipient acknowledgement', () => {
|
|
const fixture = installedFixture();
|
|
const options = deterministicOptions(fixture);
|
|
let control = openControlPlane(options);
|
|
const direct = control.sendMessage({
|
|
sender: 'session-manager', recipient: fixture.worker, body: 'Inspect the bounded change.',
|
|
references: { task: 'TASK-1', path: 'index.js' },
|
|
});
|
|
const group = control.sendMessage({ sender: fixture.worker, recipient: 'all', body: 'Shared boundary changed.' });
|
|
control.close();
|
|
|
|
control = openControlPlane(options);
|
|
const inbox = control.readMessages({ recipient: fixture.worker, unacknowledgedOnly: true });
|
|
assert.deepEqual(inbox.map((message) => message.id), [direct.id, group.id]);
|
|
assert.equal(inbox[0].projectId, 'control-plane-fixture');
|
|
assert.equal(inbox[0].references.task, 'TASK-1');
|
|
assert.equal(inbox[0].acknowledgedAt, null);
|
|
const acknowledgement = control.ackMessage({ recipient: fixture.worker, messageId: direct.id });
|
|
assert.equal(acknowledgement.recipient, fixture.worker);
|
|
assert.equal(control.readMessages({ recipient: fixture.worker, unacknowledgedOnly: true })
|
|
.some((message) => message.id === direct.id), false);
|
|
assert.throws(
|
|
() => control.ackMessage({ recipient: 'session-manager', messageId: direct.id }),
|
|
(error) => error.code === 'MESSAGE_NOT_ADDRESSABLE',
|
|
);
|
|
control.close();
|
|
});
|
|
|
|
test('SCN-control-plane-unread-observation-metadata: a bound inbox observation is complete, counted and read-only', () => {
|
|
const fixture = installedFixture();
|
|
const control = openControlPlane(deterministicOptions(fixture));
|
|
const acknowledgedHistory = Array.from({ length: 21 }, (_, index) => control.sendMessage({
|
|
sender: 'session-manager', recipient: fixture.worker, body: `Acknowledged history ${index + 1}.`,
|
|
}));
|
|
for (const message of acknowledgedHistory) control.ackMessage({ recipient: fixture.worker, messageId: message.id });
|
|
const directUnread = control.sendMessage({
|
|
sender: 'session-manager', recipient: fixture.worker, body: 'Direct unread handoff.',
|
|
});
|
|
const allUnread = control.sendMessage({
|
|
sender: 'session-manager', recipient: 'all', body: 'Shared unread handoff.',
|
|
});
|
|
const before = readFileSync(join(control.stateRoot, 'state.db'));
|
|
|
|
const firstUnread = observeWithoutDurableWrite(control, fixture,
|
|
{ recipient: fixture.worker, selection: 'unread', limit: 1 });
|
|
assert.deepEqual(firstUnread.messages.map((message) => message.id), [directUnread.id]);
|
|
assert.deepEqual(firstUnread.observation, {
|
|
selection: 'unread', requestedLimit: 1, returnedCount: 1,
|
|
complete: false, truncated: true, pendingUnreadCount: 2,
|
|
});
|
|
|
|
const completeUnread = observeWithoutDurableWrite(control, fixture,
|
|
{ recipient: fixture.worker, selection: 'unread', limit: 2 });
|
|
assert.deepEqual(completeUnread.messages.map((message) => message.id), [directUnread.id, allUnread.id]);
|
|
assert.deepEqual(completeUnread.observation, {
|
|
selection: 'unread', requestedLimit: 2, returnedCount: 2,
|
|
complete: true, truncated: false, pendingUnreadCount: 2,
|
|
});
|
|
|
|
const history = observeWithoutDurableWrite(control, fixture,
|
|
{ recipient: fixture.worker, selection: 'history', limit: 20 });
|
|
assert.deepEqual(history.messages.map((message) => message.id), acknowledgedHistory.slice(0, 20).map((message) => message.id));
|
|
assert.deepEqual(history.observation, {
|
|
selection: 'history', requestedLimit: 20, returnedCount: 20,
|
|
complete: false, truncated: true, pendingUnreadCount: 2,
|
|
});
|
|
assert.deepEqual(readFileSync(join(control.stateRoot, 'state.db')), before,
|
|
'observing messages must not acknowledge or otherwise mutate durable state');
|
|
|
|
control.ackMessage({ recipient: fixture.worker, messageId: directUnread.id });
|
|
control.ackMessage({ recipient: fixture.worker, messageId: allUnread.id });
|
|
const beforeEmpty = readFileSync(join(control.stateRoot, 'state.db'));
|
|
const empty = observeWithoutDurableWrite(control, fixture,
|
|
{ recipient: fixture.worker, selection: 'unread', limit: 1 });
|
|
assert.deepEqual(empty.messages, []);
|
|
assert.deepEqual(empty.observation, {
|
|
selection: 'unread', requestedLimit: 1, returnedCount: 0,
|
|
complete: true, truncated: false, pendingUnreadCount: 0,
|
|
});
|
|
assert.deepEqual(readFileSync(join(control.stateRoot, 'state.db')), beforeEmpty,
|
|
'an empty observation must not mutate acknowledgement, task or ownership state');
|
|
const originalObserveMessages = control.observeMessages;
|
|
control.observeMessages = function auditedReadMutant(input) {
|
|
const observed = originalObserveMessages.call(this, input);
|
|
this.audit({
|
|
actorId: input.recipient,
|
|
operation: 'diagnostic.forbidden-observation-write',
|
|
entityType: 'message-observation',
|
|
});
|
|
return observed;
|
|
};
|
|
try {
|
|
assert.throws(
|
|
() => observeWithoutDurableWrite(control, fixture,
|
|
{ recipient: fixture.worker, selection: 'unread', limit: 1 }),
|
|
(error) => /DML write execution/.test(error.message)
|
|
&& /INSERT INTO audit_events/.test(error.message),
|
|
);
|
|
} finally {
|
|
control.observeMessages = originalObserveMessages;
|
|
}
|
|
assert.equal(control.observeMessages, originalObserveMessages,
|
|
'the mutation wrapper must be restored after its required invariant failure');
|
|
control.observeMessages = function directMultilineAuditedReadMutant(input) {
|
|
const observed = originalObserveMessages.call(this, input);
|
|
this.database.prepare(`
|
|
INSERT INTO audit_events (id, project_id, actor_id, operation, entity_type, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
`).run(
|
|
'direct-multiline-diagnostic', this.projectId, input.recipient,
|
|
'diagnostic.direct-read-write', 'message-observation', '2026-10-01T12:00:00Z',
|
|
);
|
|
return observed;
|
|
};
|
|
try {
|
|
assert.throws(
|
|
() => observeWithoutDurableWrite(control, fixture,
|
|
{ recipient: fixture.worker, selection: 'unread', limit: 1 }),
|
|
(error) => /executed 1 DML write execution/.test(error.message)
|
|
&& /INSERT INTO audit_events/.test(error.message),
|
|
'a leading-whitespace direct multiline INSERT must be counted at the fixture connection boundary',
|
|
);
|
|
} finally {
|
|
control.observeMessages = originalObserveMessages;
|
|
}
|
|
assert.equal(control.observeMessages, originalObserveMessages,
|
|
'the direct multiline mutation wrapper must be restored after its required invariant failure');
|
|
assert.deepEqual(observeWithoutDurableWrite(control, fixture,
|
|
{ recipient: fixture.worker, selection: 'unread', limit: 1 }).messages, [],
|
|
'the restored original observation still passes the complete no-write invariant');
|
|
control.close();
|
|
});
|
|
|
|
test('SCN-control-plane-unread-observation-refusal: invalid observation inputs fail closed', () => {
|
|
const fixture = installedFixture();
|
|
const control = openControlPlane(deterministicOptions(fixture));
|
|
for (const selection of [undefined, 'all', 'unacknowledged']) {
|
|
assert.throws(
|
|
() => control.observeMessages({ recipient: fixture.worker, selection, limit: 1 }),
|
|
(error) => error.code === 'INVALID_MESSAGE_SELECTION',
|
|
);
|
|
}
|
|
for (const limit of [undefined, 0, 1.5, 1001]) {
|
|
assert.throws(
|
|
() => control.observeMessages({ recipient: fixture.worker, selection: 'history', limit }),
|
|
(error) => error.code === 'INVALID_MESSAGE_LIMIT',
|
|
);
|
|
}
|
|
for (const recipient of ['all', 'unregistered-recipient']) {
|
|
assert.throws(
|
|
() => control.observeMessages({ recipient, selection: 'history', limit: 1 }),
|
|
(error) => error.code === 'UNKNOWN_FLEET_IDENTITY',
|
|
);
|
|
}
|
|
control.close();
|
|
});
|
|
|
|
test('SCN-identity-presence-authority: stable Fleet identity outlives runtime IDs and forged actors are rejected', () => {
|
|
const fixture = installedFixture();
|
|
const control = openControlPlane(deterministicOptions(fixture));
|
|
control.reportStatus({
|
|
areaId: fixture.worker, state: 'working', summary: 'Implementing TASK-1',
|
|
runtime: 'claude', runtimeSessionId: 'claude-session-old', task: 'TASK-1',
|
|
});
|
|
control.reportStatus({
|
|
areaId: fixture.worker, state: 'waiting', summary: 'Runtime replaced',
|
|
runtime: 'claude', runtimeSessionId: 'claude-session-new', task: 'TASK-1',
|
|
});
|
|
const identity = control.identity(fixture.worker);
|
|
assert.equal(identity.areaId, fixture.worker);
|
|
assert.equal(identity.runtimeSessionId, 'claude-session-new');
|
|
assert.equal(identity.state, 'waiting');
|
|
assert.throws(
|
|
() => control.sendMessage({ sender: 'forged-agent', recipient: fixture.worker, body: 'Override ownership.' }),
|
|
(error) => error.code === 'UNKNOWN_FLEET_IDENTITY',
|
|
);
|
|
assert.throws(
|
|
() => control.reportStatus({ areaId: fixture.worker, state: 'invented', summary: 'Nope' }),
|
|
(error) => error.code === 'INVALID_PRESENCE_STATE',
|
|
);
|
|
control.close();
|
|
});
|
|
|
|
test('SCN-ownership-handoff: path ownership is live and handoffs do not mutate the approved roster', () => {
|
|
const fixture = installedFixture();
|
|
const control = openControlPlane(deterministicOptions(fixture));
|
|
const before = control.whoOwns({ path: 'index.js' });
|
|
assert.equal(before.owners.some((owner) => owner.areaId === fixture.worker), true);
|
|
const handoff = control.requestHandoff({
|
|
sender: fixture.worker, recipient: 'session-manager', path: 'index.js',
|
|
reason: 'Need a reviewed ownership ruling.', task: 'TASK-1',
|
|
});
|
|
assert.equal(handoff.status, 'requested');
|
|
assert.equal(control.whoOwns({ path: 'index.js' }).owners.some((owner) => owner.areaId === fixture.worker), true);
|
|
const managerInbox = control.readMessages({ recipient: 'session-manager' });
|
|
assert.equal(managerInbox.some((message) => message.references.handoff === handoff.id), true);
|
|
const coordination = control.requestCoordination({
|
|
sender: fixture.worker, participants: [fixture.qa].filter(Boolean),
|
|
body: 'Please coordinate the cross-domain verification.', task: 'TASK-1',
|
|
});
|
|
assert.equal(coordination.recipient, 'session-manager');
|
|
control.close();
|
|
});
|
|
|
|
test('SCN-local-state-path-safety: a repository-edited manifest cannot redirect SQLite writes', () => {
|
|
const fixture = installedFixture();
|
|
const manifestPath = join(fixture.root, '.torch', 'install-manifest.json');
|
|
const manifest = JSON.parse(readFileSync(manifestPath, 'utf8'));
|
|
manifest.external.find((entry) => entry.type === 'local-state').path = fixture.root;
|
|
writeFileSync(manifestPath, `${JSON.stringify(manifest, null, 2)}\n`);
|
|
assert.throws(
|
|
() => openControlPlane({ repositoryRoot: fixture.root, env: fixture.env }),
|
|
(error) => error.code === 'LOCAL_STATE_PATH_MISMATCH',
|
|
);
|
|
assert.equal(existsSync(join(fixture.root, 'state.db')), false);
|
|
});
|