Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 56 additions & 0 deletions electron/automation-destination.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
"use strict";

/** Missing mode keeps legacy new-thread-per-run definitions unchanged. */
function destinationFields(input, base = {}) {
const sessionMode = input.sessionMode === undefined
? (base.sessionMode || "new-thread") : input.sessionMode;
if (!["new-thread", "continue", "fresh"].includes(sessionMode)) {
throw new Error("Invalid automation session mode");
}
const raw = input.destinationThreadId === undefined
? base.destinationThreadId : input.destinationThreadId;
const destinationThreadId = sessionMode === "new-thread" ? null : String(raw || "").trim();
if (sessionMode !== "new-thread" && !destinationThreadId) {
throw new Error("Destination thread is required");
}
return { sessionMode, destinationThreadId };
}

/** Save validates identity and authority; execution additionally requires an idle target. */
function destinationThread(store, auto, runner, staticOnly = false) {
const { sessionMode, destinationThreadId } = destinationFields(auto);
if (sessionMode === "new-thread") return null;
const t = store.getThread(destinationThreadId);
if (!t || t.projectId !== auto.projectId) {
throw new Error("Automation destination must exist in its project");
}
if (t.archived || t.trashedAt || t.deletedAt || t.teleportedTo ||
t.memoryConsolidate || t.ask || t.pendingFork) {
throw new Error("Automation destination is unavailable");
}
if (!staticOnly && (runner?.isRunning?.(t.id) ||
t.status === "working" || t.status === "quota-wait" ||
t.quotaWaitUntil || t.quotaFailoverPending || t.pendingPlan ||
t.pendingQuestion || t.pendingApproval || t.pendingSecret || t.awaitingInput ||
t.runQueue || t.queued || t.interruptedAt || t.btw?.length ||
runner?.getPendingPermission?.(t.id))) {
throw Object.assign(
new Error("Automation destination has unfinished work or pending input; run skipped"),
{ code: "AUTOMATION_DESTINATION_BUSY" },
);
}
// Fixed destinations retain identity, but cannot lend their additional authority.
if (t.provider !== auto.provider || (t.model || null) !== (auto.model || null) ||
t.providerInstance || t.agentProfileId || t.autoApprove ||
(t.permissionMode && t.permissionMode !== "default") ||
t.folderWriteGrants?.length || t.restartGrant || t.alwaysAllowKeys?.length ||
t.alwaysAllowRules?.length || t.route || auto.tier) {
throw new Error("Automation destination must match provider/model with default permissions, no account override, grants, or routing");
}
if (!staticOnly && t.freshContextPending && sessionMode !== "fresh") {
throw new Error("Automation destination is waiting for a user message after fresh context");
}
return t;
}

module.exports = { destinationFields, destinationThread };
18 changes: 16 additions & 2 deletions electron/automations-file.js
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ const MAX_SHARED_AUTOMATIONS = 16;
const PROMPT_MAX = 4000;
const PRESETS = new Set(["hourly", "daily", "weekly", "cron", "once"]);
/** canonical()'s keys: anything else in a row is a typo (#438). */
const ROW_KEYS = new Set(["id", "name", "prompt", "provider", "model", "preset", "hour", "cron", "runAt", "trigger", "tier", "budget"]);
const ROW_KEYS = new Set(["sessionMode", "destinationThreadId", "id", "name", "prompt", "provider", "model", "preset", "hour", "cron", "runAt", "trigger", "tier", "budget"]);

/**
* Overlay / merge id for a shared automation row.
Expand Down Expand Up @@ -71,6 +71,8 @@ function slugFromName(name, taken) {
*/
function canonical(row) {
return {
sessionMode: row.sessionMode,
destinationThreadId: row.destinationThreadId,
id: row.id,
name: row.name,
prompt: row.prompt,
Expand Down Expand Up @@ -238,8 +240,12 @@ function parseRow(raw, index, seen) {
return { error: `solenta.json: ${at}.${msg}` };
}

let destination;
try { destination = require("./automation-destination.js").destinationFields(r); }
catch (err) { return { error: `solenta.json: ${at}: ${err.message}` }; }
return {
row: {
...(r.sessionMode !== undefined || r.destinationThreadId !== undefined ? destination : {}),
id,
name,
prompt,
Expand Down Expand Up @@ -338,7 +344,7 @@ function isLocalProject(project) {
* @returns {object[]}
*/
function mergeAutomations(store, nowMs) {
const { nextFire } = require("./automations.js");
const { nextFire, MAX_THREADS_PER_AUTOMATION } = require("./automations.js");
const privates = (store.getAutomations() || []).filter(
(a) => a && a.scope !== "project",
);
Expand All @@ -363,6 +369,10 @@ function mergeAutomations(store, nowMs) {
lastError: overlay.lastError,
consecutiveFailures: overlay.consecutiveFailures,
pendingRunThreadId: overlay.pendingRunThreadId,
...(overlay.pendingRunId !== undefined ? { pendingRunId: overlay.pendingRunId } : {}),
...(Array.isArray(overlay.runHistory)
? { runHistory: overlay.runHistory.slice(0, MAX_THREADS_PER_AUTOMATION) }
: {}),
nextRunAt: overlay.nextRunAt,
stage: overlay.stage,
graduate: overlay.graduate,
Expand All @@ -388,6 +398,8 @@ function mergeAutomations(store, nowMs) {
projectId: project.id,
name: row.name,
prompt: row.prompt,
sessionMode: row.sessionMode,
destinationThreadId: row.destinationThreadId,
provider: row.provider,
model: row.model,
preset: row.preset,
Expand Down Expand Up @@ -500,6 +512,8 @@ function ensureOverlay(store, auto) {
scope: "project",
name: auto.name,
prompt: auto.prompt,
sessionMode: auto.sessionMode,
destinationThreadId: auto.destinationThreadId,
provider: auto.provider,
model: auto.model,
preset: auto.preset,
Expand Down
61 changes: 51 additions & 10 deletions electron/automations.js
Original file line number Diff line number Diff line change
Expand Up @@ -304,8 +304,14 @@ function settleAutomationRuns(ctx) {
const t = ctx.store.getThread(auto.pendingRunThreadId);
if (t && (t.status === "working" || t.status === "quota-wait")) continue;
changed = true;
if (auto.pendingRunId && auto.runHistory?.length) patchAutomation(ctx.store, auto.id, {
pendingRunId: null,
runHistory: auto.runHistory.map((r) => r.runId === auto.pendingRunId ? { ...r, status: t?.status || "failed" } : r),
});
if (t) {
const last = ctx.store.getLastAssistantMessage(t.id);
const last = auto.pendingRunId
? ctx.store.getMessages(t.id).filter((m) => m.role === "assistant" && m.runId === auto.pendingRunId).at(-1)
: ctx.store.getLastAssistantMessage(t.id);
const memory = extractMemory(last && last.text);
if (memory != null) patchAutomation(ctx.store, auto.id, { memory });
}
Expand Down Expand Up @@ -368,7 +374,7 @@ function pruneAutomationThreads(store, automationId) {
indexed.sort((a, b) => (b.t.createdAt - a.t.createdAt) || (b.i - a.i));
for (let i = MAX_THREADS_PER_AUTOMATION; i < indexed.length; i++) {
const t = indexed[i].t;
if (isRetainedAutomationThread(t)) continue;
if (isRetainedAutomationThread(t) || require("./automations-file.js").mergeAutomations(store, Date.now()).some((a) => a.destinationThreadId === t.id || a.runHistory?.some((r) => r.threadId === t.id))) continue;
services.purgeThread(store, t.id);
}
}
Expand All @@ -383,7 +389,25 @@ function pruneAutomationThreads(store, automationId) {
* @param {number} now
* @param {{ rethrow?: boolean, triggerNote?: string }} [opts]
*/
const firingByStore = new WeakMap();
async function runAutomation(ctx, auto, now, opts) {
let firing = firingByStore.get(ctx.store);
if (!firing) firingByStore.set(ctx.store, firing = new Set());
const keys = [`automation:${auto.id}`];
if (auto.sessionMode && auto.sessionMode !== "new-thread") keys.push(`thread:${auto.destinationThreadId}`);
if (keys.some((key) => firing.has(key))) {
const error = new Error("Automation destination is already starting a run");
if (opts?.rethrow) throw error;
return;
}
for (const key of keys) firing.add(key);
try {
return await fireAutomation(ctx, auto, now, opts);
} finally {
for (const key of keys) firing.delete(key);
}
}
async function fireAutomation(ctx, auto, now, opts) {
const services = require("./services.js");
const nextRunAt = nextFire(auto.preset, auto.hour, now, auto.nextRunAt, auto.cron, auto.runAt);
// #388: settle ran first in the tick, so the breaker count is current.
Expand All @@ -400,7 +424,12 @@ async function runAutomation(ctx, auto, now, opts) {

let runErr = null;
try {
const thread = services.createThread(ctx.store, {
settleAutomationRuns(ctx);
const pending = ctx.store.getAutomation(auto.id)?.pendingRunThreadId;
const pendingThread = pending && ctx.store.getThread(pending);
if (auto.sessionMode && auto.sessionMode !== "new-thread" && pendingThread && (pendingThread.status === "working" || pendingThread.status === "quota-wait")) throw Object.assign(new Error("Automation still has an unfinished run; run skipped"), { code: "AUTOMATION_DESTINATION_BUSY" });
const target = require("./automation-destination.js").destinationThread(ctx.store, auto, ctx.runner);
const thread = target || services.createThread(ctx.store, {
projectId: auto.projectId,
title: auto.name,
automationId: auto.id,
Expand All @@ -409,8 +438,12 @@ async function runAutomation(ctx, auto, now, opts) {
// #286: the automation's cap is harness state on the thread it fires.
const { budgetPatchFrom } = require("./budgetCaps.js");
const capped = budgetPatchFrom(auto);
if (capped.budget) ctx.store.updateThread(thread.id, capped);
services.setProvider(ctx.store, {
if (capped.budget) {
const budget = { ...(thread.budget || {}) };
for (const [key, limit] of Object.entries(capped.budget)) budget[key] = Math.min(budget[key] ?? Infinity, limit);
ctx.store.updateThread(thread.id, { budget });
}
if (!target) services.setProvider(ctx.store, {
threadId: thread.id,
provider: auto.provider,
model: auto.model,
Expand All @@ -420,25 +453,31 @@ async function runAutomation(ctx, auto, now, opts) {
if (typeof ctx.broadcast === "function") {
ctx.broadcast("threads:changed", services.listThreads(ctx.store));
}
await ctx.runner.startRun({
const launch = await (target ? ctx.runner.startAutomationRun.bind(ctx.runner) : ctx.runner.startRun.bind(ctx.runner))({
...(target ? { automationId: auto.id } : {}),
threadId: thread.id,
prompt: buildRunPrompt({ ...auto, ...life }, opts && opts.triggerNote),
});
patchAutomation(ctx.store, auto.id, {
lastError: null,
pendingRunThreadId: thread.id,
...((target || ctx.store.getAutomation(auto.id)?.runHistory?.length) && launch?.runId ? { pendingRunId: launch.runId, runHistory: [
{ id: require("node:crypto").randomUUID(), threadId: thread.id, newThread: !target, runId: launch.runId, startedAt: now, status: "working" },
...(ctx.store.getAutomation(auto.id)?.runHistory || []),
].slice(0, MAX_THREADS_PER_AUTOMATION) } : {}),
});
ctx.store.save();
} catch (err) {
runErr = err;
recordFailure(ctx, auto.id, errorText(err));
if (err?.code === "AUTOMATION_DESTINATION_BUSY") patchAutomation(ctx.store, auto.id, { lastError: errorText(err) });
else recordFailure(ctx, auto.id, errorText(err));
ctx.store.save();
}

// #134: prune after startRun so a thrown purge cannot undo a completed fire.
// Own try/catch: retention failure must not stamp lastError or fail the run.
try {
pruneAutomationThreads(ctx.store, auto.id);
if (!auto.sessionMode || auto.sessionMode === "new-thread") pruneAutomationThreads(ctx.store, auto.id);
ctx.store.save();
const { scheduleImagePruneFromStore } = require("./image-store.js");
void scheduleImagePruneFromStore(ctx.store);
Expand Down Expand Up @@ -473,22 +512,24 @@ function listAutomationRuns(store, automationId) {
if (!auto) {
throw new Error(`Unknown automation: ${id}`);
}
const fixedRuns = (auto.runHistory || []).map(({ id, threadId, runId, startedAt, status }, i) => ({ id, threadId, runId, startedAt, status, pending: auto.pendingRunId === runId && auto.pendingRunThreadId === threadId }));
Comment thread
currentbits marked this conversation as resolved.
Resolved
const indexed = store
.getThreads()
.map((t, i) => ({ t, i }))
.filter(
({ t }) =>
t &&
t.automationId === id &&
!auto.runHistory?.some((r) => r.newThread && r.threadId === t.id) &&
t.projectId === auto.projectId &&
t.memoryConsolidate !== true,
);
indexed.sort((a, b) => b.t.createdAt - a.t.createdAt || b.i - a.i);
const runs = indexed.map(({ t }) => ({
const runs = [...fixedRuns, ...indexed.map(({ t }) => ({
threadId: t.id,
startedAt: t.createdAt,
status: t.status || "idle",
}));
}))].sort((a, b) => b.startedAt - a.startedAt);
return {
automationId: id,
runs,
Expand Down
1 change: 1 addition & 0 deletions electron/ipc-git.js
Original file line number Diff line number Diff line change
Expand Up @@ -325,6 +325,7 @@ module.exports = {
"git:prTemplate": async (_ctx, input) => {
return readPrTemplate(input && input.projectPath);
},
"git:prCompletions": async (_ctx, input) => require("./prCompletions.js").prCompletions(input),
"git:prDetail": async (_ctx, input) => {
return viewPrDetail(input && input.projectPath, input && input.prNumber, {
repo: input && input.repo,
Expand Down
6 changes: 6 additions & 0 deletions electron/orchServer.js
Original file line number Diff line number Diff line change
Expand Up @@ -1440,6 +1440,8 @@ function createToolHandlers(deps) {
}
const created = addAutomation(store, {
projectId: thread.projectId,
sessionMode: args && args.sessionMode,
destinationThreadId: args && args.destinationThreadId,
name: args && args.name,
prompt: args && args.prompt,
provider: (args && args.provider) || thread.provider,
Expand All @@ -1460,6 +1462,8 @@ function createToolHandlers(deps) {
ok: true,
id: created.id,
name: created.name,
sessionMode: created.sessionMode || "new-thread",
destinationThreadId: created.destinationThreadId || null,
projectId: created.projectId,
prompt: created.prompt,
provider: created.provider,
Expand Down Expand Up @@ -2452,6 +2456,8 @@ function buildMcpServer(sdk, handlers, opts = {}) {
inputSchema: {
threadId: z.string().min(1),
projectId: z.string().min(1),
sessionMode: z.enum(["new-thread", "continue", "fresh"]).optional(),
destinationThreadId: z.string().optional(),
name: z.string().min(1),
prompt: z.string().min(1),
provider: z.string().min(1).optional(),
Expand Down
70 changes: 70 additions & 0 deletions electron/prCompletions.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
"use strict";
const { gitTryAsync, ghApiTryAsync, tailErr } = require('./worktrees.js');
const { gitlabRemote, glabApi } = require('./gitlab.js');
const { parseRepoSlug } = require('./prLinks.js');

// Every request resolves origin anew. Never search all users or all accounts.
async function prCompletions(input) {
const cwd = String(input?.projectPath || '');
const kind = input?.kind;
if (!cwd || !['@', '#', '!'].includes(kind)) return { ok: false, reason: 'Invalid completion request' };
// Match the textarea token grammar at the trust boundary too. In particular,
// GitHub search qualifiers, spaces and quotes must never augment repo scope.
const query = input.query == null ? '' : input.query;
if (typeof query !== 'string' || !/^[\w.-]{0,100}$/.test(query)) {
return { ok: false, reason: 'Invalid completion query' };
}
const page = Math.max(1, Math.min(10, Math.trunc(Number(input.page) || 1)));
const remote = await gitTryAsync(cwd, ['remote', 'get-url', 'origin']);
if (!remote.ok) return { ok: false, reason: 'Repository unavailable' };
const origin = String(remote.stdout || '').trim();
const gl = gitlabRemote(origin);
let rows;
let exact = null;
const numeric = kind !== "@" && page === 1 && /^[1-9]\d*$/.test(query) && Number.isSafeInteger(Number(query));
if (gl) {
// Explicit cross-repo requests stay on the origin host and its authenticated account.
if (input.repo) gl.path = String(input.repo);
const endpoint = kind === '@' ? '/members/all' : kind === '!' ? '/merge_requests' : '/issues';
const [result, direct] = await Promise.all([
glabApi(cwd, gl, 'GET', `${endpoint}?per_page=20&page=${page}&${kind === '@' ? 'query' : 'search'}=${encodeURIComponent(query)}`),
numeric ? glabApi(cwd, gl, 'GET', `${endpoint}/${Number(query)}`) : null,
]);
if (direct?.ok && Number(direct.data?.iid) === Number(query)) exact = direct.data;
if (!result.ok && !exact) return result;
rows = result.ok && Array.isArray(result.data) ? result.data : [];
} else {
if (kind === '!') return { ok: true, items: [], hasMore: false };
const slug = parseRepoSlug(input.repo || origin);
if (!slug) return { ok: false, reason: 'Repository unavailable' };
const endpoint = kind === '@'
? `repos/${slug}/assignees?per_page=20&page=${page}`
: `search/issues?q=${encodeURIComponent(`repo:${slug} ${query}`)}&per_page=20&page=${page}`;
const host = origin.match(/^[^@/]+@([^:]+):/)?.[1] || (() => { try { return new URL(origin).hostname; } catch { return null; } })();
if (!host) return { ok: false, reason: 'Repository host unavailable' };
const request = endpoint => ghApiTryAsync(cwd, ['api', endpoint, '--hostname', host], { originUrl: origin, timeout: 15000 });
const [result, direct] = await Promise.all([
request(endpoint), numeric ? request(`repos/${slug}/issues/${Number(query)}`) : null,
]);
if (direct?.ok) {
try { const data = JSON.parse(direct.stdout); if (Number(data?.number) === Number(query)) exact = data; } catch { /* search remains available */ }
}
if (!result.ok && !exact) return { ok: false, reason: tailErr(result.stderr || result.combined, 'Completion search failed') };
try { const data = result.ok ? JSON.parse(result.stdout) : []; rows = Array.isArray(data) ? data : data.items || []; }
catch { if (!exact) return { ok: false, reason: 'Invalid completion response' }; rows = []; }
}
const hasMore = rows.length >= 20 && page < 10;
// Pin the exact reference without displacing any of the 20 search rows or
// shifting subsequent search pages. A first numeric page has at most 21 items.
if (exact) rows = [exact, ...rows.filter(row => Number(row.iid || row.number) !== Number(query))];
const items = rows.map(row => kind === '@'
? { value: `@${row.username || row.login}`, label: row.name || row.username || row.login }
: { value: `${kind}${row.iid || row.number}`, label: row.title || '' })
.filter(row => row.value.toLowerCase().includes(query.toLowerCase()) || row.label.toLowerCase().includes(query.toLowerCase()))
.slice(0, exact ? 21 : 20);
return { ok: true, items, hasMore };
}
module.exports = { prCompletions: async input => {
try { return await prCompletions(input); }
catch { return { ok: false, reason: 'Completion search unavailable' }; }
} };
Loading
Loading