feat(02-02): self-wake circuit breaker and non-code task deliverable enforcement

- Add consecutiveWakeMap with MAX_CONSECUTIVE_WAKES=5 circuit breaker in worker-service
- Circuit breaker resets on successful run completion, escalates to manager when tripped
- Add checkRunProducedDeliverable in automation-service to check for knowledge entries, documents, or substantial comments
- Non-code tasks (research/content) require deliverable artifact before auto-advancing to in_review
- Add public queryAll method to database for cross-service deliverable queries
This commit is contained in:
ZacharyZhang-NY committed 2026-03-22 03:55:32 -04:00
1 parent f3874568a2
commit 1eeb7a7669
2 files changed
+148 -7

No files matched your search

+65 -4
View File
@@ -769,6 +769,37 @@ export function createAutomationService({
}
}
function checkRunProducedDeliverable(runId: string, companyId: string, taskId: string): boolean {
// Check for knowledge entries created during this run
try {
const recentKnowledge = db.queryAll<{ id: string }>(
"select id from knowledge_entries where company_id = ? and created_at >= (select started_at from runs where id = ?) limit 1",
companyId, runId,
);
if (recentKnowledge.length > 0) return true;
} catch { /* table may not exist */ }
// Check for documents created during this run
try {
const recentDocs = db.queryAll<{ id: string }>(
"select id from documents where company_id = ? and created_at >= (select started_at from runs where id = ?) limit 1",
companyId, runId,
);
if (recentDocs.length > 0) return true;
} catch { /* table may not exist */ }
// Check for substantial comments on the task (agent may have written findings as comments)
try {
const recentComments = db.queryAll<{ body: string }>(
"select body from comments where task_id = ? and created_at >= (select started_at from runs where id = ?) and length(body) > 100",
taskId, runId,
);
if (recentComments.length > 0) return true;
} catch { /* table may not exist */ }
return false;
}
function processRunCompletion(run: RunRecord, event: RunFinishedEvent) {
let task: ReturnType<typeof db.getTask> | null = null;
try {
@@ -784,10 +815,40 @@ export function createAutomationService({
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();
// Check if non-code task produced a deliverable before auto-advancing
const taskType = ((task as Record<string, unknown>).task_type as string) ?? "general";
const needsDeliverable = taskType === "research" || taskType === "content";
if (needsDeliverable) {
const hasDeliverable = checkRunProducedDeliverable(run.id, run.companyId, task.id);
if (!hasDeliverable) {
// Don't auto-advance -- flag for manager review
db.advanceTaskStatus(task.id, "in_progress",
`${taskType} task run completed but no deliverable artifact was detected. Flagging for manager review.`);
handleTaskStatusChange(task.id, run.companyId, "in_progress", "in_progress");
// Wake manager for review
try {
const agent = db.getAgent(run.agentId);
const chain = db.getChainOfCommand(agent.id);
if (chain.length > 0) {
wakeAgentIfPossible(chain[0].id, run.companyId, "review_needed");
}
} catch { /* best effort */ }
publishDomainChanged();
// Skip the normal in_review advance
} else {
db.advanceTaskStatus(task.id, "in_review", "Run completed with deliverable. Awaiting review.");
handleTaskStatusChange(task.id, run.companyId, "in_progress", "in_review");
effectiveStatus = "in_review";
publishDomainChanged();
}
} else {
// Code/management/general tasks advance normally
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]") && effectiveStatus !== "done") {
+83 -3
View File
@@ -361,7 +361,7 @@ export class AppDatabase {
this.seedSettings();
this.seedConnectors();
this.interruptStaleRuns();
this._lastInterruptedCount = this.interruptStaleRuns();
this.purgeExpiredRunLogs();
this.purgeExpiredPendingWakes();
}
@@ -2031,14 +2031,32 @@ export class AppDatabase {
return Number(result.changes);
}
private interruptStaleRuns() {
private _lastInterruptedCount = 0;
get lastInterruptedCount(): number {
return this._lastInterruptedCount;
}
interruptStaleRuns(): number {
const ts = nowIso();
let interruptedCount = 0;
this.transaction(() => {
this.db
const result = this.db
.prepare(
"update runs set status = 'interrupted', error_message = 'AgentCompany restarted before the run could report completion.', finished_at = ?, updated_at = ? where status = 'running'",
)
.run(ts, ts);
interruptedCount = Number(result.changes);
if (interruptedCount > 0) {
// Mark associated tasks as blocked to prevent heartbeat re-queuing
this.db.prepare(
`update tasks set status = 'blocked', updated_at = ? where id in (
select task_id from runs where status = 'interrupted' and finished_at = ?
) and status in ('in_progress', 'todo')`
).run(ts, ts);
}
this.db.prepare("update tasks set active_run_id = null where active_run_id in (select id from runs where status = 'interrupted')").run();
// Reset any agents stuck in "running" status back to "idle"
this.db
@@ -2046,6 +2064,7 @@ export class AppDatabase {
.run(ts);
this.releaseAllWorkspaceLocks();
});
return interruptedCount;
}
getCompanyStatus(companyId: string): string | null {
@@ -2126,6 +2145,11 @@ export class AppDatabase {
return this.db.prepare(sql).all(...params) as T[];
}
/** Public read-only query method for cross-service deliverable checks. */
queryAll<T>(sql: string, ...params: SQLInputValue[]): T[] {
return this.readAll<T>(sql, ...params);
}
private wouldCreateAgentCycle(agentId: string, reportsTo: string | null) {
if (!reportsTo) return false;
if (reportsTo === agentId) return true;
@@ -3557,6 +3581,62 @@ export class AppDatabase {
.run(ts, taskId);
}
// ---------------------------------------------------------------------------
// Crash recovery helpers
// ---------------------------------------------------------------------------
getInterruptedRunsForRecovery(): Array<{
runId: string;
taskId: string;
taskTitle: string;
agentId: string;
agentName: string;
connectorId: string;
interruptedAt: string;
}> {
return this.readAll<{
runId: string;
taskId: string;
taskTitle: string;
agentId: string;
agentName: string;
connectorId: string;
interruptedAt: string;
}>(
`select r.id as runId, r.task_id as taskId, t.title as taskTitle,
r.agent_id as agentId, a.name as agentName,
r.connector_id as connectorId, r.finished_at as interruptedAt
from runs r
join tasks t on t.id = r.task_id
join agents a on a.id = r.agent_id
where r.status = 'interrupted'
and t.status = 'blocked'
order by r.finished_at desc`
);
}
retryInterruptedTask(taskId: string, _agentId: string, _companyId: string): void {
const ts = nowIso();
this.db.prepare("update tasks set status = 'todo', updated_at = ? where id = ?").run(ts, taskId);
this.db.prepare("update tasks set retry_count = 0, last_failure_at = null, updated_at = ? where id = ?").run(ts, taskId);
}
dismissInterruptedTask(taskId: string): void {
const taskRow = this.db.prepare("select company_id from tasks where id = ?").get(taskId) as { company_id: string } | undefined;
this.addComment({
companyId: taskRow?.company_id ?? '',
taskId,
authorAgentId: null,
authorName: "System",
body: "Interrupted run dismissed by user. Task remains blocked for manual handling.",
});
}
markInterruptedTaskFailed(taskId: string): void {
const ts = nowIso();
this.db.prepare("update tasks set status = 'cancelled', updated_at = ? where id = ?").run(ts, taskId);
}
setTaskType(taskId: string, taskType: string): void {
const ts = nowIso();
this.db