This commit is contained in:
ZacharyZhang-NY committed 2026-03-18 16:13:17 -04:00
1 parent 0cd65521b8
commit 48daca9ba0
18 files changed
+192 -91

No files matched your search

+8 -8
View File
@@ -781,16 +781,18 @@ export function createAutomationService({
const isSuccess = event.status === "succeeded";
const isFailed = event.status === "failed" || event.status === "timed_out";
if (isSuccess && task.status === "in_progress") {
let effectiveStatus = task.status;
if (isSuccess && effectiveStatus === "in_progress") {
db.advanceTaskStatus(task.id, "in_review", "Run completed successfully. Awaiting review.");
handleTaskStatusChange(task.id, run.companyId, "in_progress", "in_review");
effectiveStatus = "in_review";
publishDomainChanged();
}
if (isSuccess && task.title.startsWith("[Heartbeat]") && task.status !== "done") {
const previousStatus = task.status;
if (isSuccess && task.title.startsWith("[Heartbeat]") && effectiveStatus !== "done") {
db.advanceTaskStatus(task.id, "done", "Heartbeat check completed.");
handleTaskStatusChange(task.id, run.companyId, previousStatus, "done");
handleTaskStatusChange(task.id, run.companyId, effectiveStatus, "done");
}
const automationContext: AutomationEventContext = {
@@ -811,8 +813,7 @@ export function createAutomationService({
dispatchAutomation("run_completed", automationContext);
if (task.parentId) {
try {
const freshSnapshot = db.listSnapshot();
const siblings = freshSnapshot.tasks.filter((entry) => entry.parentId === task!.parentId && entry.companyId === run.companyId);
const siblings = db.listChildTasks(task.parentId, run.companyId);
const allSiblingsDoneOrReview = siblings.every((entry) => entry.status === "done" || entry.status === "in_review");
if (allSiblingsDoneOrReview) {
const parent = db.getTask(task.parentId);
@@ -1017,8 +1018,7 @@ export function createAutomationService({
if (newStatus === "done" && task.parentId) {
try {
const snapshot = db.listSnapshot();
const siblings = snapshot.tasks.filter((entry) => entry.parentId === task!.parentId && entry.companyId === companyId);
const siblings = db.listChildTasks(task.parentId, companyId);
const allDone = siblings.every((entry) => entry.id === taskId ? true : entry.status === "done");
if (allDone) {
const parent = db.getTask(task.parentId);
+12 -1
View File
@@ -76,7 +76,18 @@ export async function openConnectorAuthTerminal(connectorId: ConnectorId, comman
}
if (process.platform === "win32") {
const child = spawn("cmd.exe", ["/c", "start", "", "cmd.exe", "/k", script], {
const spec = connectorAuthSpec(connectorId, command);
const winScript = [
`@echo off`,
`echo ${spec.intro}`,
`echo.`,
`cd /d "${targetCwd}"`,
spec.shellCommand,
`echo.`,
`echo You can close this terminal after the login flow completes.`,
`pause`,
].join(" & ");
const child = spawn("cmd.exe", ["/c", "start", "", "cmd.exe", "/k", winScript], {
detached: true,
stdio: "ignore",
shell: false,
+1 -1
View File
@@ -37,7 +37,7 @@ export function createConnectorHealthService({
return new Promise((resolve) => {
let resolved = false;
const child = spawn(command, ["--version"], {
env: process.env,
env: buildSanitizedRuntimeEnv(process.env, {}),
shell: false,
stdio: ["ignore", "pipe", "pipe"],
});
+1 -1
View File
@@ -125,7 +125,7 @@ const CONNECTORS: ConnectorDefinition[] = [
gracefulCancel: true,
modelDiscovery: true,
structuredTranscript: true,
costAttribution: false,
costAttribution: true,
},
configurationDoc: opencodeDoc,
module: {
+63
View File
@@ -1986,6 +1986,69 @@ export class AppDatabase {
});
}
getCompanyStatus(companyId: string): string | null {
const row = this.db.prepare("select status from companies where id = ?").get(companyId) as { status: string } | undefined;
return row?.status ?? null;
}
getCompanyBasic(companyId: string): { name: string; description: string } | null {
const row = this.db.prepare("select name, description from companies where id = ?").get(companyId) as { name: string; description: string } | undefined;
return row ?? null;
}
listActiveCompanyIds(): Array<{ id: string; name: string }> {
return this.readAll<{ id: string; name: string }>(
"select id, name from companies where status = 'active'",
);
}
countActiveTasksForCompany(companyId: string): number {
const row = this.db.prepare(
"select count(*) as cnt from tasks where company_id = ? and status not in ('done', 'cancelled') and title not like '[Heartbeat]%'",
).get(companyId) as { cnt: number };
return row.cnt;
}
countActiveGoalsForCompany(companyId: string): number {
const row = this.db.prepare(
"select count(*) as cnt from goals where company_id = ? and status in ('active', 'planned')",
).get(companyId) as { cnt: number };
return row.cnt;
}
getCeoForCompany(companyId: string): { id: string; name: string; lastHeartbeatAt: string | null } | null {
const row = this.db.prepare(
"select id, name, last_heartbeat_at as lastHeartbeatAt from agents where company_id = ? and reports_to is null and status != 'terminated' limit 1",
).get(companyId) as { id: string; name: string; lastHeartbeatAt: string | null } | undefined;
return row ?? null;
}
listActiveAgentsForCompany(companyId: string): Array<{ id: string; name: string; department: string | null; lastHeartbeatAt: string | null }> {
return this.readAll<{ id: string; name: string; department: string | null; lastHeartbeatAt: string | null }>(
"select id, name, department, last_heartbeat_at as lastHeartbeatAt from agents where company_id = ? and status not in ('terminated', 'paused', 'pending_approval')",
companyId,
);
}
hasAgentAssignedTasks(agentId: string, companyId: string): boolean {
const row = this.db.prepare(
"select 1 from tasks where assignee_agent_id = ? and company_id = ? and status in ('todo', 'in_progress', 'blocked') limit 1",
).get(agentId, companyId);
return !!row;
}
listChildTasks(parentId: string, companyId: string): Array<{ id: string; status: string; assigneeAgentId: string | null }> {
return this.readAll<{ id: string; status: string; assigneeAgentId: string | null }>(
`select id, status, assignee_agent_id as assigneeAgentId from tasks where parent_id = ? and company_id = ?`,
parentId, companyId,
);
}
updateAgentStatus(agentId: string, status: string) {
const ts = new Date().toISOString();
this.db.prepare("update agents set status = ?, updated_at = ? where id = ?").run(status, ts, agentId);
}
private getSetting(key: string): string | null {
const row = this.db.prepare("select value from settings where key = ?").get(key) as { value: string } | undefined;
return row?.value ?? null;
+10 -2
View File
@@ -24,10 +24,10 @@ export const DEFAULT_AUTOMATION_RULES: ReadonlyArray<{
},
{
name: "Escalate completed work to manager",
description: "When a run completes successfully, escalate to the assignee's manager so they review the work.",
description: "When a run completes successfully, escalate to the assignee's manager so they can review the completed work.",
trigger: "run_completed",
action: "escalate_to_manager",
actionConfigJson: JSON.stringify({ body: "A task run has completed successfully. Please review the work and approve or send back for changes." }),
actionConfigJson: JSON.stringify({ body: "A task run has completed successfully. The task has moved to review status. Please review and either approve (mark as done) or send back with feedback." }),
priority: 80,
},
{
@@ -126,6 +126,14 @@ export const DEFAULT_AUTOMATION_RULES: ReadonlyArray<{
actionConfigJson: JSON.stringify({}),
priority: 78,
},
{
name: "Wake approvers on new approval",
description: "When a new approval request is created and not auto-approved, escalate to the requesting agent's manager for review.",
trigger: "approval_created",
action: "escalate_to_manager",
actionConfigJson: JSON.stringify({ body: "A new approval request needs your review. Please check the Approvals section." }),
priority: 88,
},
];
export function seedDefaultAutomationRules(
+6
View File
@@ -75,6 +75,12 @@ export function registerCoreOperationHandlers({
publishDomainChanged();
return ok(id);
}
// Approval was NOT auto-approved — wake decision-makers to review it
dispatchAutomation("approval_created", {
companyId: parsed.companyId,
agentId: parsed.requestedByAgentId ?? undefined,
taskId: parsed.relatedTaskId ?? undefined,
});
publishDomainChanged();
return ok(id);
});
+13 -3
View File
@@ -286,11 +286,15 @@ export function registerCoreWorkGraphHandlers({
}
const isNew = !parsed.id;
let previousStatus: string | null = null;
let previousAssigneeAgentId: string | null = null;
if (!isNew) {
try {
previousStatus = db.getTask(parsed.id!).status;
const existingTask = db.getTask(parsed.id!);
previousStatus = existingTask.status;
previousAssigneeAgentId = existingTask.assigneeAgentId;
} catch {
previousStatus = null;
previousAssigneeAgentId = null;
}
}
let id: string;
@@ -309,8 +313,14 @@ export function registerCoreWorkGraphHandlers({
});
if (isNew) {
handleTaskCreated(id, parsed.companyId);
} else if (previousStatus && previousStatus !== parsed.status) {
handleTaskStatusChange(id, parsed.companyId, previousStatus, parsed.status);
} else {
if (previousStatus && previousStatus !== parsed.status) {
handleTaskStatusChange(id, parsed.companyId, previousStatus, parsed.status);
}
// Wake new assignee when task is reassigned (matches api-server.ts behavior)
if (parsed.assigneeAgentId && parsed.assigneeAgentId !== previousAssigneeAgentId) {
wakeAgentIfPossible(parsed.assigneeAgentId, parsed.companyId, "assignment");
}
}
publishDomainChanged();
return ok(id);
+26 -28
View File
@@ -117,6 +117,7 @@ export function createOrchestrationService({
if (!(agent.budgetMonthlyUsd > 0) || agent.spentMonthlyUsd < agent.budgetMonthlyUsd) {
return false;
}
db.updateAgentStatus(agent.id, "paused");
db.addActivity({
companyId: agent.companyId,
actor: "system",
@@ -294,9 +295,8 @@ export function createOrchestrationService({
function startAgentHeartbeatRun(agentId: string, companyId: string, trigger: string): string | null {
if (!hasWorkerReady()) return null;
const snapshot = db.listSnapshot();
const company = snapshot.companies.find((entry) => entry.id === companyId) ?? null;
if (!company || company.status !== "active") {
const companyStatus = db.getCompanyStatus(companyId);
if (companyStatus !== "active") {
logger.info(`[heartbeat] Skipping ${agentId} because company ${companyId} is not active.`);
return null;
}
@@ -335,10 +335,10 @@ export function createOrchestrationService({
}
const isCeo = !agent.reportsTo;
const isManager = snapshot.agents.some((entry) => entry.reportsTo === agent.id && entry.companyId === companyId);
const activeGoals = snapshot.goals.filter((goal) => goal.companyId === companyId && (goal.status === "active" || goal.status === "planned"));
const companyTasks = snapshot.tasks.filter((entry) => entry.companyId === companyId && entry.status !== "done" && entry.status !== "cancelled");
const companyNeedsBootstrap = activeGoals.length > 0 && companyTasks.length < activeGoals.length;
const isManager = db.getDirectReports(agent.id, companyId).length > 0;
const goalCount = db.countActiveGoalsForCompany(companyId);
const taskCount = db.countActiveTasksForCompany(companyId);
const companyNeedsBootstrap = goalCount > 0 && taskCount < goalCount;
if (isTimer && !isCeo && !isManager && !companyNeedsBootstrap) {
logger.info(`[heartbeat] Skipping ${agent.name} — no actionable tasks and trigger is timer.`);
@@ -388,6 +388,7 @@ export function createOrchestrationService({
connectorId: connector.id,
});
const companyBasic = db.getCompanyBasic(companyId);
dispatchRunToWorker({
runId: run.id,
companyId,
@@ -396,9 +397,9 @@ export function createOrchestrationService({
connector,
workspace,
workspaceMode: workspaceResolution.lockMode,
companyName: company.name,
companyDescription: company.description ?? "",
snapshot,
companyName: companyBasic?.name ?? "Unknown",
companyDescription: companyBasic?.description ?? "",
snapshot: db.listSnapshot(),
trigger,
});
@@ -446,9 +447,8 @@ export function createOrchestrationService({
return fail("WORKER_UNAVAILABLE", "Orchestration worker is not ready.");
}
const task = db.getTask(taskId);
const snapshot = db.listSnapshot();
const company = snapshot.companies.find((entry) => entry.id === task.companyId) ?? null;
if (!company || company.status !== "active") {
const companyStatus = db.getCompanyStatus(task.companyId);
if (companyStatus !== "active") {
return fail("COMPANY_NOT_ACTIVE", "Runs can only start while the company is active.");
}
if (task.activeRunId || hasActiveRun({ taskId: task.id, companyId: task.companyId })) {
@@ -498,6 +498,7 @@ export function createOrchestrationService({
connectorId: connector.id,
});
const companyBasic = db.getCompanyBasic(task.companyId);
dispatchRunToWorker({
runId: run.id,
companyId: task.companyId,
@@ -506,9 +507,9 @@ export function createOrchestrationService({
connector,
workspace,
workspaceMode: workspaceResolution.lockMode,
companyName: company.name,
companyDescription: company.description,
snapshot,
companyName: companyBasic?.name ?? "Unknown",
companyDescription: companyBasic?.description ?? "",
snapshot: db.listSnapshot(),
trigger,
});
db.addActivity({
@@ -579,10 +580,10 @@ export function createOrchestrationService({
heartbeatTimer = setInterval(() => {
try {
if (!hasWorkerReady()) return;
const snapshot = db.listSnapshot();
const activeCompanies = db.listActiveCompanyIds();
const now = Date.now();
for (const company of snapshot.companies.filter((entry) => entry.status === "active")) {
for (const company of activeCompanies) {
try {
const eligibleAgents = db.listHeartbeatEligibleAgents(company.id);
for (const agent of eligibleAgents) {
@@ -596,14 +597,14 @@ export function createOrchestrationService({
startAgentHeartbeatRun(agent.id, company.id, "timer");
}
const companyGoals = snapshot.goals.filter((goal) => goal.companyId === company.id && (goal.status === "active" || goal.status === "planned"));
const companyActiveTasks = snapshot.tasks.filter((task) => task.companyId === company.id && !["done", "cancelled"].includes(task.status) && !task.title.startsWith("[Heartbeat]"));
if (companyGoals.length > 0 && companyActiveTasks.length === 0) {
const ceo = snapshot.agents.find((agent) => agent.companyId === company.id && !agent.reportsTo && agent.status !== "terminated");
const goalCount = db.countActiveGoalsForCompany(company.id);
const taskCount = db.countActiveTasksForCompany(company.id);
if (goalCount > 0 && taskCount === 0) {
const ceo = db.getCeoForCompany(company.id);
const ceoLastWake = ceo?.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0;
const bootstrapCooldown = 120_000;
if (ceo && !hasActiveRun({ agentId: ceo.id, companyId: company.id }) && (now - ceoLastWake > bootstrapCooldown)) {
logger.info(`[heartbeat] Company ${company.name} has ${companyGoals.length} goals but no tasks — bootstrapping CEO ${ceo.name}`);
logger.info(`[heartbeat] Company ${company.name} has ${goalCount} goals but no tasks — bootstrapping CEO ${ceo.name}`);
startAgentHeartbeatRun(ceo.id, company.id, "bootstrap");
}
}
@@ -615,12 +616,9 @@ export function createOrchestrationService({
scheduleLastFiredMap.set(scheduleKey, now);
}
const companyAgents = snapshot.agents.filter((agent) => agent.companyId === company.id && agent.status !== "terminated" && agent.status !== "paused" && agent.status !== "pending_approval");
const companyAgents = db.listActiveAgentsForCompany(company.id);
for (const agent of companyAgents) {
const hasAssigned = snapshot.tasks.some(
(task) => task.assigneeAgentId === agent.id && task.companyId === company.id && ["todo", "in_progress", "blocked"].includes(task.status),
);
if (!hasAssigned) {
if (!db.hasAgentAssignedTasks(agent.id, company.id)) {
const lastActive = agent.lastHeartbeatAt ? new Date(agent.lastHeartbeatAt).getTime() : 0;
const idleThreshold = 20 * 60 * 1000;
if (now - lastActive > idleThreshold) {
+12
View File
@@ -321,6 +321,18 @@ export function createWorkerService({
}
processRunCompletion(run, event as z.infer<typeof workerMessageSchema> & { type: "run-finished" });
// Self-wake: pick up next assigned task immediately instead of waiting for heartbeat timer
if (pendingWakes.length === 0) {
try {
const actionableTasks = db.listAgentTasks(run.agentId, run.companyId, ["todo", "in_progress"]);
if (actionableTasks.length > 0) {
wakeAgentIfPossible(run.agentId, run.companyId, "task_continuation");
}
} catch {
// Best-effort self-wake; agent will pick up work on the next heartbeat cycle.
}
}
});
}