Add fixture-only execution materialization and refresh (#1500)
This commit is contained in:
@@ -0,0 +1,96 @@
|
||||
// Fixture-only execution resolution. Not a membership/registration authority.
|
||||
import { validId, validateProvider, validateAccount, validateSettingsProfile,
|
||||
validateHarnessManifest } from './records.mjs';
|
||||
export class FixtureError extends Error {
|
||||
constructor(code) { super(code); this.name = 'FixtureError'; this.code = code; }
|
||||
}
|
||||
export const refuse = code => { throw new FixtureError(code); };
|
||||
const obj = v => v !== null && typeof v === 'object' && !Array.isArray(v);
|
||||
export function exact(v, required, optional = []) {
|
||||
if (!obj(v) || required.some(k => !Object.hasOwn(v, k)) ||
|
||||
Object.keys(v).some(k => ![...required, ...optional].includes(k))) refuse('invalid-fixture-input');
|
||||
}
|
||||
function checkRegistry(result) {
|
||||
if (!obj(result) || !Array.isArray(result.errors) || result.errors.length || !obj(result.entries)) refuse('invalid-registry');
|
||||
const entries = result.entries;
|
||||
for (const [name, validate] of [['providers', validateProvider], ['accounts', validateAccount],
|
||||
['profiles', validateSettingsProfile], ['harnesses', validateHarnessManifest]]) {
|
||||
if (!obj(entries[name])) refuse('invalid-registry');
|
||||
for (const [key, record] of Object.entries(entries[name])) {
|
||||
if (validate(record).length) refuse('invalid-registry');
|
||||
if (key !== (name === 'accounts' ? `${record.provider}/${record.id}` : record.id)) refuse('invalid-registry');
|
||||
}
|
||||
}
|
||||
return entries;
|
||||
}
|
||||
export function resolveFixtureExecution(result, request) {
|
||||
// Clone immediately: later caller mutation cannot change this resolution.
|
||||
let r, registry;
|
||||
try { r = structuredClone(request); registry = structuredClone(result); }
|
||||
catch { refuse('invalid-fixture-input'); }
|
||||
const entries = checkRegistry(registry);
|
||||
exact(r, ['fixtureOnly', 'agentId', 'projectId', 'workspaceId', 'sessionId', 'executionId', 'profile'], ['accounts', 'fork']);
|
||||
if (r.fixtureOnly !== true) refuse('fixture-only');
|
||||
for (const key of ['agentId', 'projectId', 'workspaceId', 'sessionId', 'executionId', 'profile'])
|
||||
if (!validId(r[key])) refuse('invalid-scope');
|
||||
if (!Object.hasOwn(entries.profiles, r.profile)) refuse('missing-profile');
|
||||
const profile = entries.profiles[r.profile];
|
||||
const overrides = r.accounts ?? {};
|
||||
if (!obj(overrides)) refuse('invalid-selection');
|
||||
const providers = new Set([...(profile.providers ?? []), ...profile.allowedAccounts.map(ref => ref.split('/')[0])]);
|
||||
for (const key of Object.keys(overrides)) if (!providers.has(key)) refuse('provider-not-enrolled');
|
||||
if (Object.hasOwn(r, 'fork')) {
|
||||
exact(r.fork, ['sourceSessionId', 'accounts']);
|
||||
if (!validId(r.fork.sourceSessionId) || r.fork.sourceSessionId === r.sessionId || !obj(r.fork.accounts)) refuse('invalid-fork-pin');
|
||||
for (const id of Object.keys(r.fork.accounts)) if (!providers.has(id)) refuse('revoked-fork-pin');
|
||||
}
|
||||
const accounts = Object.create(null), models = { providers: Object.create(null) }, types = Object.create(null);
|
||||
for (const id of [...providers].sort()) {
|
||||
if (!Object.hasOwn(entries.providers, id)) refuse('missing-provider');
|
||||
const provider = entries.providers[id], cfg = provider.harnesses.pi;
|
||||
if (!cfg || !entries.harnesses.pi) refuse('unsupported-harness');
|
||||
let ref;
|
||||
if (r.fork) {
|
||||
ref = r.fork.accounts[id];
|
||||
if (Object.hasOwn(overrides, id) && overrides[id] !== ref) refuse('fork-account-change');
|
||||
} else ref = Object.hasOwn(overrides, id) ? overrides[id] : profile.defaultAccounts?.[id];
|
||||
if (ref === undefined) {
|
||||
if (!provider.credentialTypes.includes('none')) refuse(r.fork ? 'missing-fork-pin' : 'missing-account-selection');
|
||||
} else {
|
||||
if (typeof ref !== 'string' || !profile.allowedAccounts.includes(ref) || ref.split('/')[0] !== id)
|
||||
refuse('account-not-enrolled');
|
||||
if (!Object.hasOwn(entries.accounts, ref)) refuse('missing-account');
|
||||
const account = entries.accounts[ref];
|
||||
if (account.provider !== id || !provider.credentialTypes.includes(account.type)) refuse('credential-type-not-supported');
|
||||
accounts[id] = ref; types[id] = account.type;
|
||||
}
|
||||
if (provider.kind === 'native') {
|
||||
// No aliasing to another provider's credential slot.
|
||||
if (cfg.providerId !== id) refuse('provider-alias-unsupported');
|
||||
if (profile.models?.[id]) refuse('native-model-enforcement-unavailable');
|
||||
} else {
|
||||
if (!['openai-completions', 'openai-responses', 'anthropic-messages', 'google-generative-ai'].includes(cfg.api)) refuse('unsupported-model-api');
|
||||
const selected = profile.models?.[id] ?? cfg.models;
|
||||
if (!selected.length || selected.some(m => !cfg.models.includes(m))) refuse('model-not-enrolled');
|
||||
models.providers[id] = { api: cfg.api, baseUrl: cfg.baseUrl, models: selected.map(model => ({ id: model })) };
|
||||
if (!ref || types[id] === 'none') models.providers[id].apiKey = 'FIXTURE_NONE';
|
||||
}
|
||||
}
|
||||
const scope = Object.fromEntries(['agentId', 'projectId', 'workspaceId', 'sessionId', 'executionId'].map(k => [k, r[k]]));
|
||||
return { fixtureOnly: true, scope, profile: r.profile, accounts, types, models,
|
||||
fork: r.fork ? { sourceSessionId: r.fork.sourceSessionId, accounts: { ...accounts } } : null };
|
||||
}
|
||||
// Real keys/tokens are deliberately outside this slice's accepted input language.
|
||||
const marker = v => typeof v === 'string' && /^FIXTURE_[A-Z0-9_-]{1,128}$/.test(v);
|
||||
export function validateFixtureCredential(value, type) {
|
||||
if (type === 'none') { if (value !== null) refuse('invalid-fixture-credential'); return null; }
|
||||
if (type === 'api_key') {
|
||||
exact(value, ['type', 'key']);
|
||||
if (value.type !== type || !marker(value.key)) refuse('invalid-fixture-credential');
|
||||
} else if (type === 'oauth') {
|
||||
exact(value, ['type', 'access', 'refresh', 'expires']);
|
||||
if (value.type !== type || !marker(value.access) || !marker(value.refresh) ||
|
||||
!Number.isSafeInteger(value.expires) || value.expires < 0) refuse('invalid-fixture-credential');
|
||||
} else refuse('unsupported-fixture-credential');
|
||||
return structuredClone(value);
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
// In-memory synthetic store only; no filesystem or production backend.
|
||||
import { refuse, validateFixtureCredential } from './execution.mjs';
|
||||
const stores = new WeakSet();
|
||||
export const isFixtureStore = store => stores.has(store);
|
||||
export function createFixtureCredentialStore(initial) {
|
||||
const records = new Map(), locks = new Set();
|
||||
for (const [ref, value] of Object.entries(initial)) {
|
||||
if (!/^[a-z0-9][a-z0-9._-]{0,63}\/[a-z0-9][a-z0-9._-]{0,63}$/.test(ref)) refuse('invalid-account-ref');
|
||||
records.set(ref, validateFixtureCredential(value, value?.type ?? 'none'));
|
||||
}
|
||||
const store = Object.freeze({
|
||||
fixtureOnly: true,
|
||||
async transaction(refs, callback) {
|
||||
const keys = [...new Set(refs)].sort();
|
||||
for (const key of keys) {
|
||||
if (!records.has(key)) refuse('missing-fixture-credential');
|
||||
if (locks.has(key)) refuse('credential-busy');
|
||||
}
|
||||
keys.forEach(key => locks.add(key));
|
||||
try {
|
||||
const draft = new Map(keys.map(key => [key, structuredClone(records.get(key))]));
|
||||
const result = await callback(draft);
|
||||
for (const key of keys) validateFixtureCredential(draft.get(key), records.get(key)?.type ?? 'none');
|
||||
keys.forEach(key => records.set(key, structuredClone(draft.get(key))));
|
||||
return result;
|
||||
} finally { keys.forEach(key => locks.delete(key)); }
|
||||
},
|
||||
});
|
||||
stores.add(store);
|
||||
return store;
|
||||
}
|
||||
@@ -2,3 +2,6 @@ import { validateProvider, validateAccount, validateSettingsProfile, validateSea
|
||||
import { loadRegistry } from "./registry.mjs";
|
||||
|
||||
export { validateProvider, validateAccount, validateSettingsProfile, validateSeatSelection, validateHarnessManifest, ValidationError, loadRegistry };
|
||||
export { resolveFixtureExecution, FixtureError } from './execution.mjs';
|
||||
export { createFixtureCredentialStore } from './fixture-store.mjs';
|
||||
export { createFixtureWorkspace } from './materialize-fixture.mjs';
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
// Synthetic-only publication simulator. No caller-selected filesystem root,
|
||||
// production credential adapter, real Pi runner, service or live activation.
|
||||
import { mkdtemp, mkdir, open, rename, rm, lstat } from 'node:fs/promises';
|
||||
import { constants } from 'node:fs';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { FixtureError, exact, refuse, resolveFixtureExecution, validateFixtureCredential } from './execution.mjs';
|
||||
import { isFixtureStore } from './fixture-store.mjs';
|
||||
import { refreshFixtureCredential, validateRefreshOptions } from './refresh-fixture.mjs';
|
||||
const anchor = h => `/proc/self/fd/${h.fd}`;
|
||||
async function openDir(path) { return open(path, constants.O_RDONLY | constants.O_DIRECTORY | constants.O_NOFOLLOW); }
|
||||
async function verifyJson(dir, name, value) {
|
||||
const expected = Buffer.from(JSON.stringify(value, null, 2) + '\n');
|
||||
const h = await open(`${anchor(dir)}/${name}`, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
|
||||
try {
|
||||
const s = await h.stat();
|
||||
if (!s.isFile() || s.uid !== process.getuid() || (s.mode & 0o777) !== 0o600 || s.size !== expected.length) refuse('generation-validation-failed');
|
||||
const actual = Buffer.alloc(expected.length + 1);
|
||||
let size = 0;
|
||||
while (size < actual.length) {
|
||||
const { bytesRead } = await h.read(actual, size, actual.length - size, null);
|
||||
if (!bytesRead) break;
|
||||
size += bytesRead;
|
||||
}
|
||||
if (size !== expected.length || !actual.subarray(0, size).equals(expected)) refuse('generation-validation-failed');
|
||||
} finally { await h.close(); }
|
||||
}
|
||||
async function writeJson(dir, name, value) {
|
||||
const content = JSON.stringify(value, null, 2) + '\n';
|
||||
if (Buffer.byteLength(content) > 1024 * 1024) refuse('generation-too-large');
|
||||
const h = await open(`${anchor(dir)}/${name}`, constants.O_WRONLY | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, 0o600);
|
||||
try { await h.writeFile(content); await h.sync(); }
|
||||
finally { await h.close(); }
|
||||
}
|
||||
export async function createFixtureWorkspace() {
|
||||
if (process.platform !== 'linux') refuse('unsupported-platform');
|
||||
const path = await mkdtemp('/tmp/mosaic-materialization-fixture-');
|
||||
const root = await openDir(path), identity = await root.stat();
|
||||
let closed = false, active = 0;
|
||||
const handles = {};
|
||||
try {
|
||||
for (const name of ['claims', 'pending', 'generations']) {
|
||||
await mkdir(`${anchor(root)}/${name}`, { mode: 0o700 });
|
||||
handles[name] = await openDir(`${anchor(root)}/${name}`);
|
||||
}
|
||||
} catch (e) {
|
||||
for (const h of Object.values(handles)) await h.close();
|
||||
await root.close(); await rm(path, { recursive: true, force: true }); throw e;
|
||||
}
|
||||
return Object.freeze({
|
||||
fixtureOnly: true, path,
|
||||
async generate(registry, request, store, options = {}) {
|
||||
if (closed) refuse('workspace-closed');
|
||||
if (!isFixtureStore(store)) refuse('fixture-store-required');
|
||||
exact(options, [], ['fault', 'refresh']);
|
||||
if (options.fault !== undefined && !['after-auth', 'before-publish', 'after-publish'].includes(options.fault)) refuse('invalid-fixture-option');
|
||||
options = { fault: options.fault, refresh: options.refresh === undefined ? undefined : validateRefreshOptions(options.refresh) };
|
||||
const plan = resolveFixtureExecution(registry, request), id = plan.scope.executionId;
|
||||
const pending = `${anchor(handles.pending)}/${id}`, claimPath = `${anchor(handles.claims)}/${id}`;
|
||||
let claim, generation, claimed = false, published = false;
|
||||
active++;
|
||||
try {
|
||||
try { await mkdir(claimPath, { mode: 0o700 }); claimed = true; }
|
||||
catch { refuse('execution-already-claimed'); }
|
||||
claim = await openDir(claimPath);
|
||||
await writeJson(claim, 'started.json', { fixtureOnly: true, executionId: id, state: 'started' });
|
||||
const result = await store.transaction(Object.values(plan.accounts), async draft => {
|
||||
const auth = Object.create(null);
|
||||
for (const [provider, ref] of Object.entries(plan.accounts)) {
|
||||
let credential = validateFixtureCredential(draft.get(ref), plan.types[provider]);
|
||||
if (credential !== null && (options.refresh !== undefined ||
|
||||
(credential.type === 'oauth' && credential.expires <= Date.now() + 300000))) {
|
||||
credential = await refreshFixtureCredential(provider, credential, options.refresh ?? {});
|
||||
draft.set(ref, credential);
|
||||
}
|
||||
if (credential !== null) auth[provider] = credential;
|
||||
}
|
||||
await mkdir(pending, { mode: 0o700 }); generation = await openDir(pending);
|
||||
await writeJson(generation, 'auth.json', auth);
|
||||
if (options.fault === 'after-auth') refuse('injected-generation-failure');
|
||||
await writeJson(generation, 'models.json', plan.models);
|
||||
const manifest = { manifestVersion: 1, fixtureOnly: true, manifestId: randomUUID(),
|
||||
scope: plan.scope, profile: plan.profile, accounts: plan.accounts, fork: plan.fork,
|
||||
artifacts: ['auth.json', 'models.json'], state: 'ready' };
|
||||
// Manifest contains no token, expiry or credential-content digest.
|
||||
await writeJson(generation, 'manifest.json', manifest);
|
||||
if (options.fault === 'before-publish') refuse('injected-generation-failure');
|
||||
await verifyJson(generation, 'auth.json', auth);
|
||||
await verifyJson(generation, 'models.json', plan.models);
|
||||
await verifyJson(generation, 'manifest.json', manifest);
|
||||
await generation.sync(); await generation.close(); generation = undefined;
|
||||
const target = `${anchor(handles.generations)}/${id}`;
|
||||
try { await lstat(target); refuse('generation-exists'); }
|
||||
catch (e) { if (e.code !== 'ENOENT') throw e; }
|
||||
// Retained exclusive claim serializes this execution ID, including
|
||||
// failed attempts. Published generations are never replaced by API.
|
||||
await rename(pending, target); published = true;
|
||||
if (options.fault === 'after-publish') refuse('injected-generation-failure');
|
||||
await handles.generations.sync();
|
||||
await writeJson(claim, 'result.json', { fixtureOnly: true, state: 'published', manifestId: manifest.manifestId });
|
||||
return { fixtureOnly: true, manifest, generationPath: `${path}/generations/${id}` };
|
||||
});
|
||||
return result;
|
||||
} catch (e) {
|
||||
if (generation) { await generation.close(); generation = undefined; }
|
||||
if (claimed && !published) await rm(pending, { recursive: true, force: true });
|
||||
const code = published ? 'publication-uncertain' : e instanceof FixtureError ? e.code : 'generation-failed';
|
||||
if (claim) {
|
||||
try { await writeJson(claim, 'failure.json', { fixtureOnly: true, state: published ? 'uncertain' : 'failed', code }); }
|
||||
catch { throw new FixtureError('recording-failed'); }
|
||||
}
|
||||
throw new FixtureError(code);
|
||||
} finally {
|
||||
if (claim) await claim.close(); active--;
|
||||
}
|
||||
},
|
||||
async close() {
|
||||
if (closed) return;
|
||||
if (active) refuse('workspace-busy');
|
||||
closed = true;
|
||||
for (const h of Object.values(handles)) await h.close();
|
||||
await root.close();
|
||||
const current = await lstat(path);
|
||||
if (current.dev !== identity.dev || current.ino !== identity.ino || current.isSymbolicLink()) refuse('workspace-path-changed');
|
||||
await rm(path, { recursive: true, force: true });
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
// Deliberately executes a fixed fake program, never Pi or caller-supplied code.
|
||||
// PI_CODING_AGENT_DIR matches the statically verified 0.85.1 auth boundary.
|
||||
import { spawn } from 'node:child_process';
|
||||
import { mkdtemp, mkdir, writeFile, open, rm } from 'node:fs/promises';
|
||||
import { constants } from 'node:fs';
|
||||
import { validId } from './records.mjs';
|
||||
import { exact, refuse, validateFixtureCredential } from './execution.mjs';
|
||||
const fake = `
|
||||
import { readFile, writeFile } from 'node:fs/promises';
|
||||
import { join } from 'node:path';
|
||||
const [command, subcommand, flag, provider] = process.argv.slice(1);
|
||||
if (command !== 'auth' || subcommand !== 'check' || flag !== '--provider') process.exit(2);
|
||||
if (Object.keys(process.env).some(k => !['HOME','PI_CODING_AGENT_DIR','FIXTURE_MODE'].includes(k))) process.exit(2);
|
||||
const path = join(process.env.PI_CODING_AGENT_DIR, 'auth.json');
|
||||
const mode = process.env.FIXTURE_MODE;
|
||||
if (mode === 'timeout') await new Promise(() => setInterval(() => {}, 1000));
|
||||
if (mode === 'failure') { process.stderr.write('FIXTURE_PRIVATE_DIAGNOSTIC'); process.exit(1); }
|
||||
if (mode === 'malformed') { await writeFile(path, '{', { mode: 0o600 }); process.exit(0); }
|
||||
const auth = JSON.parse(await readFile(path, 'utf8'));
|
||||
const c = auth[provider];
|
||||
if (mode === 'rotate' && c.type === 'oauth') {
|
||||
c.access = 'FIXTURE_ROTATED_ACCESS'; c.refresh = 'FIXTURE_ROTATED_REFRESH';
|
||||
c.expires = Date.now() + 3600000;
|
||||
}
|
||||
await writeFile(path, JSON.stringify(auth), { mode: 0o600 });
|
||||
process.stdout.write('ready');
|
||||
`;
|
||||
export function validateRefreshOptions(options = {}) {
|
||||
exact(options, [], ['mode', 'timeoutMs']);
|
||||
const mode = options.mode ?? 'rotate', timeoutMs = options.timeoutMs ?? 2000;
|
||||
if (!['rotate', 'unchanged', 'failure', 'timeout', 'malformed'].includes(mode) ||
|
||||
!Number.isInteger(timeoutMs) || timeoutMs < 10 || timeoutMs > 10000) refuse('invalid-refresh-option');
|
||||
return { mode, timeoutMs };
|
||||
}
|
||||
export async function refreshFixtureCredential(provider, credential, options = {}) {
|
||||
if (!validId(provider)) refuse('invalid-provider');
|
||||
const { mode, timeoutMs } = validateRefreshOptions(options);
|
||||
const input = validateFixtureCredential(credential, credential?.type);
|
||||
const root = await mkdtemp('/tmp/mosaic-refresh-fixture-');
|
||||
try {
|
||||
const agent = `${root}/agent`, home = `${root}/home`, cwd = `${root}/cwd`;
|
||||
for (const dir of [agent, home, cwd]) await mkdir(dir, { mode: 0o700 });
|
||||
const file = `${agent}/auth.json`;
|
||||
await writeFile(file, JSON.stringify({ [provider]: input }), { mode: 0o600, flag: 'wx' });
|
||||
const outcome = await new Promise(resolve => {
|
||||
let timedOut = false, spawnFailed = false;
|
||||
const child = spawn(process.execPath, ['--input-type=module', '-e', fake, 'auth', 'check', '--provider', provider], {
|
||||
cwd, env: { HOME: home, PI_CODING_AGENT_DIR: agent, FIXTURE_MODE: mode },
|
||||
// Child output is discarded, never buffered, parsed, logged or returned.
|
||||
stdio: 'ignore', shell: false,
|
||||
});
|
||||
const timer = setTimeout(() => { timedOut = true; child.kill('SIGKILL'); }, timeoutMs);
|
||||
child.on('error', () => { spawnFailed = true; });
|
||||
child.on('close', (code, signal) => {
|
||||
clearTimeout(timer); resolve({ code, signal, timedOut, spawnFailed });
|
||||
});
|
||||
});
|
||||
if (outcome.timedOut) refuse('refresh-timeout');
|
||||
if (outcome.spawnFailed || outcome.code !== 0 || outcome.signal) refuse('refresh-failed');
|
||||
const h = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
|
||||
let result;
|
||||
try {
|
||||
const s = await h.stat();
|
||||
if (!s.isFile() || s.uid !== process.getuid() || (s.mode & 0o777) !== 0o600 || s.size > 4096) refuse('invalid-refresh-output');
|
||||
const buf = Buffer.alloc(4097); let size = 0;
|
||||
while (size < buf.length) {
|
||||
const { bytesRead } = await h.read(buf, size, buf.length - size, null);
|
||||
if (!bytesRead) break;
|
||||
size += bytesRead;
|
||||
}
|
||||
if (size > 4096) refuse('invalid-refresh-output');
|
||||
try { result = JSON.parse(buf.subarray(0, size).toString('utf8')); }
|
||||
catch { refuse('invalid-refresh-output'); }
|
||||
} finally { await h.close(); }
|
||||
exact(result, [provider]);
|
||||
const output = validateFixtureCredential(result[provider], input.type);
|
||||
if (output.type === 'oauth' && output.expires <= Date.now() + 300000) refuse('refresh-not-ready');
|
||||
return output;
|
||||
} finally { await rm(root, { recursive: true, force: true }); }
|
||||
}
|
||||
Reference in New Issue
Block a user