surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Merge v3-t79: extension loader deadline 120 s with one retry on timeout (T79)

Browse Source
This commit is contained in:
surma
2026-11-08 05:56:15 +00:00
parent 3339e18485 420066952c
commit fcfaff6018
17 changed files with 456 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
host/js/automations-test-bundle.js
+20 -3
View File
@@ -32708,6 +32708,22 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -32802,8 +32818,8 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -32857,7 +32873,7 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -32901,6 +32917,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/bundle.js
+20 -3
View File
@@ -70716,6 +70716,22 @@ async function loadExtensions(harness2, registry3, storage2, models2, context, b
let state2 = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state2.error = message;
ops2.op_extension_status(JSON.stringify(state2));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state2), context);
@@ -70810,8 +70826,8 @@ async function loadExtensions(harness2, registry3, storage2, models2, context, b
try {
await fs2.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs2.list(`${snapshot}/extensions`).catch((error) => {
@@ -70865,7 +70881,7 @@ async function loadExtensions(harness2, registry3, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state2.error = String(error);
ops2.op_extension_status(JSON.stringify(state2));
@@ -70909,6 +70925,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/core-ext-test-bundle.js
+20 -3
View File
@@ -31482,6 +31482,22 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -31576,8 +31592,8 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -31631,7 +31647,7 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -31675,6 +31691,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/dream-test-bundle.js
+20 -3
View File
@@ -30542,6 +30542,22 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -30636,8 +30652,8 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -30691,7 +30707,7 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -30735,6 +30751,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/events-test-bundle.js
+20 -3
View File
@@ -31482,6 +31482,22 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -31576,8 +31592,8 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -31631,7 +31647,7 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -31675,6 +31691,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/extensions-test-bundle.js
+20 -3
View File
@@ -31482,6 +31482,22 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -31576,8 +31592,8 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -31631,7 +31647,7 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -31675,6 +31691,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/extensions.ts
+20 -3
View File
@@ -83,6 +83,23 @@ export async function loadExtensions(harness:any, registry:any, storage:any, mod
let state = {...(await harness.snapshot(ExtensionState,context) ?? {good:null,current:null,error:null})};
let settings:any = {};
let active = true;
const commands = new AbortController();
// Snapshot checkout copies the source tree. The tool helper's 30 seconds is too short under load.
const loaderGit = async (...args:string[]) => {
for (let attempt=0;;attempt++) {
try {
const result = await exec(['git',...args],{cwd:repo,timeout:120,signal:commands.signal});
if(result.code)throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch(error) {
const message = `git ${args.join(' ')}: ${String(error)}`;
state.error=message;
ops.op_extension_status(JSON.stringify(state));
if(attempt===0 && !commands.signal.aborted && String(error).includes('command timed out'))continue;
throw new Error(message);
}
}
};
let queue = Promise.resolve();
const publish = async () => {
await harness.commit(async (tx:any)=>Object.assign(await tx.doc(ExtensionState),state),context);
@@ -134,7 +151,7 @@ export async function loadExtensions(harness:any, registry:any, storage:any, mod
try {
const snapshot = `/run/sausage/extensions/${sha}`;
for(const path of ['/run/sausage','/run/sausage/extensions']) {try {await fs.mkdir(path);} catch(error) {if((error as any).code!=='already_exists')throw error;} }
try {await fs.stat(snapshot);} catch {await git(repo,'worktree','prune');await git(repo,'worktree','add','--detach',snapshot,sha);}
try {await fs.stat(snapshot);} catch {await loaderGit('worktree','prune');await loaderGit('worktree','add','--detach',snapshot,sha);}
const modules:any[] = [];
for(const entry of await fs.list(`${snapshot}/extensions`).catch(error=>{if(error.code==='not_found')return [];throw error;})) {
if(entry.kind!=='directory')continue;
@@ -164,7 +181,7 @@ export async function loadExtensions(harness:any, registry:any, storage:any, mod
};
const reload = async () => {
let tip:string;
try {tip=await git(repo,'rev-parse','refs/heads/main');} catch(error) {state.error=String(error);ops.op_extension_status(JSON.stringify(state));return;}
try {tip=await loaderGit('rev-parse','refs/heads/main');} catch(error) {state.error=String(error);ops.op_extension_status(JSON.stringify(state));return;}
try {settings=JSON.parse(await fs.read(`${repo}/settings.json`));} catch(error) {if((error as any).code==='not_found')settings={};else {state.notice=`Settings: ${String(error)}`;ops.op_extension_status(JSON.stringify(state));ops.op_output(`${state.notice}\n`);}}
if(state.current===tip && facets.length)return;
state.current=tip;
@@ -185,5 +202,5 @@ export async function loadExtensions(harness:any, registry:any, storage:any, mod
const action = facet?.actions?.[key];
if (!action) throw Object.assign(new Error('Unknown user action'),{status:404});
return action(args,{host:facet.host,signal,conversationId,source:'ui'});
}, close:async()=>{active=false;Deno.core.close(rid);await pending;await queue;await dispose(facets);}};
}, close:async()=>{active=false;commands.abort();Deno.core.close(rid);await pending;await queue;await dispose(facets);}};
}
host/js/goal-questions-test-bundle.js
+20 -3
View File
@@ -32708,6 +32708,22 @@ async function loadExtensions(harness, registry, storage, models, context, built
let state = { ...await harness.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -32802,8 +32818,8 @@ async function loadExtensions(harness, registry, storage, models, context, built
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -32857,7 +32873,7 @@ async function loadExtensions(harness, registry, storage, models, context, built
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -32901,6 +32917,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/harness-api-test-bundle.js
+20 -3
View File
@@ -31482,6 +31482,22 @@ async function loadExtensions(harness2, registry, storage, models, context, buil
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -31576,8 +31592,8 @@ async function loadExtensions(harness2, registry, storage, models, context, buil
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -31631,7 +31647,7 @@ async function loadExtensions(harness2, registry, storage, models, context, buil
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -31675,6 +31691,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/messages-test-bundle.js
+20 -3
View File
@@ -30542,6 +30542,22 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -30636,8 +30652,8 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
try {
await fs.stat(snapshot2);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot2, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot2, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot2}/extensions`).catch((error) => {
@@ -30691,7 +30707,7 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -30735,6 +30751,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/p3-test-bundle.js
+20 -3
View File
@@ -30542,6 +30542,22 @@ async function loadExtensions(harness, registry, storage, models, context, built
let state = { ...await harness.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -30636,8 +30652,8 @@ async function loadExtensions(harness, registry, storage, models, context, built
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -30691,7 +30707,7 @@ async function loadExtensions(harness, registry, storage, models, context, built
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -30735,6 +30751,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/plan-todos-test-bundle.js
+20 -3
View File
@@ -31482,6 +31482,22 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -31576,8 +31592,8 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -31631,7 +31647,7 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -31675,6 +31691,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/subagents-test-bundle.js
+20 -3
View File
@@ -30542,6 +30542,22 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -30636,8 +30652,8 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
try {
await fs.stat(snapshot2);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot2, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot2, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot2}/extensions`).catch((error) => {
@@ -30691,7 +30707,7 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -30735,6 +30751,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/watchdog-test-bundle.js
+20 -3
View File
@@ -30542,6 +30542,22 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -30636,8 +30652,8 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha2);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha2);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -30691,7 +30707,7 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -30735,6 +30751,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/js/workspaces-test-bundle.js
+20 -3
View File
@@ -30542,6 +30542,22 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
let state = { ...await harness2.snapshot(ExtensionState, context) ?? { good: null, current: null, error: null } };
let settings = {};
let active = true;
const commands = new AbortController();
const loaderGit = async (...args) => {
for (let attempt = 0; ; attempt++) {
try {
const result = await exec(["git", ...args], { cwd: repo, timeout: 120, signal: commands.signal });
if (result.code) throw new Error(result.stderr || `git exited ${result.code}`);
return result.stdout.trim();
} catch (error) {
const message = `git ${args.join(" ")}: ${String(error)}`;
state.error = message;
ops2.op_extension_status(JSON.stringify(state));
if (attempt === 0 && !commands.signal.aborted && String(error).includes("command timed out")) continue;
throw new Error(message);
}
}
};
let queue2 = Promise.resolve();
const publish2 = async () => {
await harness2.commit(async (tx) => Object.assign(await tx.doc(ExtensionState), state), context);
@@ -30636,8 +30652,8 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
try {
await fs.stat(snapshot);
} catch {
await git(repo, "worktree", "prune");
await git(repo, "worktree", "add", "--detach", snapshot, sha);
await loaderGit("worktree", "prune");
await loaderGit("worktree", "add", "--detach", snapshot, sha);
}
const modules = [];
for (const entry2 of await fs.list(`${snapshot}/extensions`).catch((error) => {
@@ -30691,7 +30707,7 @@ async function loadExtensions(harness2, registry2, storage2, models2, context, b
const reload = async () => {
let tip;
try {
tip = await git(repo, "rev-parse", "refs/heads/main");
tip = await loaderGit("rev-parse", "refs/heads/main");
} catch (error) {
state.error = String(error);
ops2.op_extension_status(JSON.stringify(state));
@@ -30735,6 +30751,7 @@ Last good: ${fallback}` : failed;
return action(args, { host: facet.host, signal, conversationId, source: "ui" });
}, close: async () => {
active = false;
commands.abort();
Deno.core.close(rid);
await pending;
await queue2;
host/tests/it/extensions.rs
+133
View File
@@ -112,3 +112,136 @@ async fn commits_and_restart() -> anyhow::Result<()> {
.await?;
Ok(())
}
/// Model command time without a wall-clock delay, including work longer than the old 30 second budget.
struct LoaderCommand {
root: Arc<dyn sausage_vfs::Layer>,
failures: usize,
attempts: std::sync::atomic::AtomicUsize,
timeout: bool,
}
#[async_trait::async_trait]
impl Exec for LoaderCommand {
async fn exec(
&self,
command: Command,
options: Options,
output: &mut (dyn for<'a> FnMut(Stream, &'a [u8]) + Send),
) -> anyhow::Result<i32> {
let Command::Argv(args) = command else {
anyhow::bail!("unexpected shell")
};
match args[1].as_str() {
"rev-parse" => output(Stream::Stdout, b"new\n"),
"worktree" if args[2] == "prune" => {}
"worktree" if args[2] == "add" => {
let attempt = self
.attempts
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
if options
.timeout
.is_some_and(|budget| budget < std::time::Duration::from_secs(31))
|| (attempt < self.failures && self.timeout)
{
anyhow::bail!("command timed out");
}
if attempt < self.failures {
anyhow::bail!("checkout failed");
}
if attempt > 0 {
let status = self
.root
.open("/proc/extensions", Default::default())
.await?;
let status = String::from_utf8(status.read(0, 65536).await?)?;
assert!(
status.contains("command timed out"),
"Retry must publish its error: {status}"
);
}
let snapshot = &args[4];
for path in [
snapshot.clone(),
format!("{snapshot}/extensions"),
format!("{snapshot}/extensions/demo"),
] {
self.root.mkdir(&path).await?;
}
let file = self
.root
.open(
&format!("{snapshot}/extensions/demo/server.ts"),
sausage_vfs::Open {
write: true,
create: true,
..Default::default()
},
)
.await?;
file.write(0, b"export default {name:'demo'};".to_vec())
.await?;
}
_ => anyhow::bail!("unexpected command {args:?}"),
}
Ok(0)
}
}
async fn loader_command(failures: usize, timeout: bool) -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let world = World::open(dir.path())?;
std::fs::create_dir_all(dir.path().join("world/src/sausage/.git"))?;
let exec = Arc::new(LoaderCommand {
root: world.root.clone(),
failures,
timeout,
attempts: std::sync::atomic::AtomicUsize::new(0),
});
let (library, _) = include_str!("../../js/bundle.js")
.split_once("// main.ts")
.unwrap();
let expected = if failures > 1 || !timeout && failures > 0 {
"old"
} else {
"new"
};
let source = format!(
r#"{library}
const testRegistry = createRegistry();
const testStorage = await openStorage(globalThis.storagePath);
const testHarness = await Harness.open(testStorage, {{models:createModels(),registry:testRegistry}}, BACKGROUND_CONTEXT);
await testHarness.root(BACKGROUND_CONTEXT);
await testHarness.commit(async tx => Object.assign(await tx.doc(ExtensionState), {{good:'old',current:'old',error:null}}), BACKGROUND_CONTEXT);
for (const path of ['/run/sausage','/run/sausage/extensions','/run/sausage/extensions/old','/run/sausage/extensions/old/extensions','/run/sausage/extensions/old/extensions/demo']) await fs2.mkdir(path);
await fs2.write('/run/sausage/extensions/old/extensions/demo/server.ts', "export default {{name:'demo'}};");
const testLoader = await loadExtensions(testHarness,testRegistry,testStorage,createModels(),BACKGROUND_CONTEXT);
const testState = testLoader.state();
if(testState.good !== '{expected}') throw new Error(JSON.stringify(testState));
if((testState.error !== null) !== ({expected:?} === 'old')) throw new Error(JSON.stringify(testState));
if(testState.error && !testState.error.includes('git worktree add')) throw new Error(testState.error);
await testLoader.close();
await testHarness.close(BACKGROUND_CONTEXT);
"#
);
sausage::harness::run_module(world, exec.clone(), &source).await?;
assert_eq!(
exec.attempts.load(std::sync::atomic::Ordering::SeqCst),
if timeout && failures > 0 { 2 } else { 1 }
);
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn slow_loader_command() -> anyhow::Result<()> {
loader_command(0, true).await
}
#[tokio::test(flavor = "current_thread")]
async fn loader_timeout_retries_once() -> anyhow::Result<()> {
loader_command(1, true).await
}
#[tokio::test(flavor = "current_thread")]
async fn loader_timeout_keeps_last_good() -> anyhow::Result<()> {
loader_command(2, true).await
}
#[tokio::test(flavor = "current_thread")]
async fn loader_real_failure_is_not_retried() -> anyhow::Result<()> {
loader_command(1, false).await
}
host/tests/it/p1.rs
+23 -1
View File
@@ -286,6 +286,28 @@ async fn bundle_copy_and_real_shell() -> anyhow::Result<()> {
Ok(())
}
/// Preserve command timings in captured test output when the loaded-machine regression fails.
struct LoaderTiming(std::sync::Arc<dyn sausage::exec::Exec>);
#[async_trait::async_trait]
impl sausage::exec::Exec for LoaderTiming {
async fn exec(
&self,
command: sausage::exec::Command,
options: sausage::exec::Options,
output: &mut (dyn for<'a> FnMut(sausage::exec::Stream, &'a [u8]) + Send),
) -> anyhow::Result<i32> {
let label = format!("{command:?}");
let budget = options.timeout;
let started = std::time::Instant::now();
let result = self.0.exec(command, options, output).await;
eprintln!(
"Loader {label}: deadline {budget:?}, elapsed {:?}, result {result:?}",
started.elapsed()
);
result
}
}
#[tokio::test(flavor = "current_thread")]
async fn bundled_extensions_load() -> anyhow::Result<()> {
let bundle = bundle()?;
@@ -320,7 +342,7 @@ async fn bundled_extensions_load() -> anyhow::Result<()> {
await bundleHarness.close(BACKGROUND_CONTEXT);
"#
);
run_module(world, exec, &source).await?;
run_module(world, std::sync::Arc::new(LoaderTiming(exec)), &source).await?;
let status = root.open("/proc/extensions", Open::default()).await?;
let status: serde_json::Value = serde_json::from_slice(&status.read(0, 65536).await?)?;
let commit =