This commit is contained in:
ZacharyZhang-NY committed 2026-04-22 13:19:43 -04:00
1 parent aba2dac3c5
commit dffeccf3e2
17 files changed
+1296 -120

No files matched your search

+11 -1
View File
@@ -1,6 +1,6 @@
import { getEnvApiKey } from "@mariozechner/pi-ai";
import type { AppDatabase } from "./database";
import { getConnectorDefinition } from "./connectors";
import { getConnectorDefinition, parseEnvBindingText } from "./connectors";
import type { ConnectorId } from "@shared/types";
import type { CredentialStore } from "./credentials";
@@ -54,6 +54,16 @@ export function createConnectorHealthService({
if (definition.provider && getEnvApiKey(definition.provider as never)) {
return { status: "ready", authState: "ok", lastError: null };
}
// Re-parse envBindingText on every probe: it is user-editable and must
// not be cached into credentialStore. An explicit binding for the
// connector's declared envKey (non-empty value) counts as ready.
const connector = db.connectorsCrud.findConnector(id);
const boundValue = connector
? parseEnvBindingText(connector.envBindingText)[definition.auth.envKey]
: undefined;
if (boundValue && boundValue.length > 0) {
return { status: "ready", authState: "ok", lastError: null };
}
return {
status: "auth_required",
authState: "required",
+262 -87
View File
@@ -1,3 +1,4 @@
import { powerMonitor } from "electron";
import { logAppError } from "../errors/index.js";
import type { HeartbeatStartResult } from "./heartbeat";
import type { OrchestrationDependencies } from "./types";
@@ -21,13 +22,44 @@ export function createHeartbeatScheduler(
* plus the first skip-and-record cycle re-establish backoff.
*/
const attemptedAtByAgent = new Map<string, number>();
/**
* Tracks consecutive skip streak per agent so the per-turn prompt
* builder (which reads `runtime_state_json` at dispatch time) can
* surface "skipped: worker_not_ready (last N ticks since T)" and stop
* being blind to a chain of zero-delivery heartbeats. Keyed by agent
* id → { reason, count, since }. Reset when status flips to `started`
* or the reason changes.
*/
const skipStreakByAgent = new Map<string, { reason: string; count: number; since: string }>();
function persistSkip(agentId: string, companyId: string, reason: string) {
/**
* Periodic heartbeat.skip_ongoing emission gate. The activity table
* would otherwise show a 5-day gap of blankness while a stall
* persists, because `writeSkipActivity` used to fire only on
* first-seen + reason-change. Every SKIP_ONGOING_EVERY_N skips of the
* same reason we emit one `heartbeat.skip_ongoing` row so the audit
* timeline carries the running count. At 15s ticks, N=40 ≈ once
* every ~10 minutes of sustained skipping — readable, not floody.
*/
const SKIP_ONGOING_EVERY_N = 40;
function persistSkip(
agentId: string,
companyId: string,
reason: string,
streak: { count: number; since: string },
lastTickAt: string,
) {
try {
const existing = db.agents.getRuntimeState(agentId) as Record<string, unknown>;
db.agents.setRuntimeState(agentId, companyId, {
...existing,
heartbeatSkip: { reason, at: new Date().toISOString() },
heartbeatSkip: {
reason,
streak: streak.count,
since: streak.since,
lastTickAt,
},
});
} catch (error) {
logAppError(logger, error, {
@@ -56,101 +88,244 @@ export function createHeartbeatScheduler(
}
}
function tryWake(agentId: string, companyId: string, agentName: string, trigger: string) {
attemptedAtByAgent.set(agentId, Date.now());
const result = startAgentHeartbeatRun(agentId, companyId, trigger);
if (result.status === "skipped") {
persistSkip(agentId, companyId, result.reason);
logger.info(`[heartbeat] ${agentName} not ready: ${result.reason}`);
} else {
clearSkip(agentId, companyId);
}
}
function start() {
if (timer) clearInterval(timer);
timer = setInterval(() => {
/**
* Structured skip write to the activity stream. Emits on first skip,
* on reason change, and every SKIP_ONGOING_EVERY_N consecutive skips
* of the same reason. That periodic row stops the "5 days of blank
* activity" pathology — after the initial `heartbeat.skipped` the
* next visible row used to be the flip/resume, which meant long
* sleep/standby stalls were invisible in the audit feed.
*
* `detail` is emitted as JSON `{reason, streak, since, trigger,
* agentName}` so downstream readers parse structured fields instead
* of regexing English. The prompt-side consumer reads
* `runtime_state_json.heartbeatSkip` (written by `persistSkip`), so
* activity here is pure audit.
*/
function writeSkipActivity(
agentId: string,
companyId: string,
agentName: string,
reason: string,
trigger: string,
tickAt: string,
): { count: number; since: string } {
const previous = skipStreakByAgent.get(agentId);
const sameReason = previous && previous.reason === reason;
const nextStreak = sameReason
? { reason, count: previous.count + 1, since: previous.since }
: { reason, count: 1, since: tickAt };
skipStreakByAgent.set(agentId, nextStreak);
const action = !sameReason
? "heartbeat.skipped"
: nextStreak.count % SKIP_ONGOING_EVERY_N === 0
? "heartbeat.skip_ongoing"
: null;
if (action) {
try {
if (isQuitting() || !hasWorkerReady()) return;
const activeCompanies = db.company.listActiveCompanyIds();
const now = Date.now();
for (const company of activeCompanies) {
try {
const eligibleAgents = db.agents.listHeartbeatEligibleAgents(company.id);
for (const agent of eligibleAgents) {
if (hasActiveRun({ agentId: agent.id, companyId: company.id })) continue;
// Unified co-founder mode pause: checkpoint engine checks mode + wait states + pending approvals
if (deps.checkpointEngine?.shouldPauseHeartbeat(company.id, agent.id)) {
continue;
}
// Use the max of last-success and last-attempt so a chain
// of skipped attempts still cools down the cycle instead
// of retrying on every 15-second tick.
const lastHeartbeat = agent.lastHeartbeatAt ? new Date(agent.lastHeartbeatAt).getTime() : 0;
const lastAttempt = attemptedAtByAgent.get(agent.id) ?? 0;
const lastTouch = Math.max(lastHeartbeat, lastAttempt);
const intervalMs = agent.heartbeatIntervalSec * 1000;
if (now - lastTouch < intervalMs) continue;
tryWake(agent.id, company.id, agent.name, "timer");
}
const goalCount = db.entityCrud.countActiveGoalsForCompany(company.id);
// Bootstrap when the company has active goals but no work is
// actually advancing. The previous gate ("taskCount === 0")
// missed the case where every task is in_review or blocked —
// the company looks busy to the counter but produces zero
// progress, and the CEO would not be woken until the next
// regular timer interval. countAdvancingTasksForCompany
// counts only todo + in_progress so bootstrapping still
// fires once the team stalls.
const advancingCount = db.tasks.countAdvancingTasksForCompany(company.id);
if (goalCount > 0 && advancingCount === 0) {
const ceo = db.agents.getCeoForCompany(company.id);
const ceoLastWake = ceo?.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0;
const ceoLastAttempt = ceo ? attemptedAtByAgent.get(ceo.id) ?? 0 : 0;
const ceoLastTouch = Math.max(ceoLastWake, ceoLastAttempt);
const bootstrapCooldown = 120_000;
if (ceo && !hasActiveRun({ agentId: ceo.id, companyId: company.id }) && (now - ceoLastTouch > bootstrapCooldown)) {
logger.info(`[heartbeat] Company ${company.name} has ${goalCount} goal(s) and no advancing tasks — bootstrapping CEO ${ceo.name}`);
tryWake(ceo.id, company.id, ceo.name, "bootstrap");
}
}
const scheduleKey = `schedule-${company.id}`;
const lastScheduleFired = scheduleLastFiredMap.get(scheduleKey) ?? 0;
if (now - lastScheduleFired >= 60_000) {
dispatchAutomation("schedule", { companyId: company.id });
scheduleLastFiredMap.set(scheduleKey, now);
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.heartbeat.company_tick_failed",
message: `Heartbeat tick failed for company ${company.id}`,
context: { companyId: company.id },
});
}
}
db.activityMod.addActivity({
companyId,
actor: "system",
action,
entityType: "agent",
entityId: agentId,
detail: JSON.stringify({
agentName,
reason,
streak: nextStreak.count,
since: nextStreak.since,
trigger,
}),
});
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.heartbeat.tick_failed",
message: "Heartbeat scheduler tick raised; next interval will retry",
});
code: "scheduler.heartbeat.activity_write_failed",
message: `Failed to append ${action} activity for ${agentName}`,
context: { agentId, companyId, reason },
}, "warn");
}
}, 15_000);
}
return { count: nextStreak.count, since: nextStreak.since };
}
function clearSkipStreak(agentId: string, companyId: string, agentName: string) {
const previous = skipStreakByAgent.get(agentId);
if (!previous) return;
skipStreakByAgent.delete(agentId);
try {
db.activityMod.addActivity({
companyId,
actor: "system",
action: "heartbeat.skip_cleared",
entityType: "agent",
entityId: agentId,
detail: JSON.stringify({
agentName,
reason: previous.reason,
streak: previous.count,
since: previous.since,
resumedAt: new Date().toISOString(),
}),
});
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.heartbeat.activity_write_failed",
message: `Failed to append heartbeat skip-cleared activity for ${agentName}`,
context: { agentId, companyId },
}, "warn");
}
}
function tryWake(agentId: string, companyId: string, agentName: string, trigger: string) {
const tickAt = new Date().toISOString();
attemptedAtByAgent.set(agentId, Date.now());
const result = startAgentHeartbeatRun(agentId, companyId, trigger);
if (result.status === "skipped") {
const streak = writeSkipActivity(agentId, companyId, agentName, result.reason, trigger, tickAt);
persistSkip(agentId, companyId, result.reason, streak, tickAt);
logger.info(`[heartbeat] ${agentName} not ready: ${result.reason} (streak=${streak.count})`);
} else {
clearSkip(agentId, companyId);
clearSkipStreak(agentId, companyId, agentName);
}
}
function runTick() {
try {
if (isQuitting() || !hasWorkerReady()) return;
const activeCompanies = db.company.listActiveCompanyIds();
const now = Date.now();
for (const company of activeCompanies) {
try {
const eligibleAgents = db.agents.listHeartbeatEligibleAgents(company.id);
for (const agent of eligibleAgents) {
if (hasActiveRun({ agentId: agent.id, companyId: company.id })) continue;
// Unified co-founder mode pause: checkpoint engine checks mode + wait states + pending approvals
if (deps.checkpointEngine?.shouldPauseHeartbeat(company.id, agent.id)) {
continue;
}
// Use the max of last-success and last-attempt so a chain
// of skipped attempts still cools down the cycle instead
// of retrying on every 15-second tick.
const lastHeartbeat = agent.lastHeartbeatAt ? new Date(agent.lastHeartbeatAt).getTime() : 0;
const lastAttempt = attemptedAtByAgent.get(agent.id) ?? 0;
const lastTouch = Math.max(lastHeartbeat, lastAttempt);
const intervalMs = agent.heartbeatIntervalSec * 1000;
if (now - lastTouch < intervalMs) continue;
tryWake(agent.id, company.id, agent.name, "timer");
}
const goalCount = db.entityCrud.countActiveGoalsForCompany(company.id);
// Bootstrap when the company has active goals but no work is
// actually advancing. The previous gate ("taskCount === 0")
// missed the case where every task is in_review or blocked —
// the company looks busy to the counter but produces zero
// progress, and the CEO would not be woken until the next
// regular timer interval. countAdvancingTasksForCompany
// counts only todo + in_progress so bootstrapping still
// fires once the team stalls.
const advancingCount = db.tasks.countAdvancingTasksForCompany(company.id);
if (goalCount > 0 && advancingCount === 0) {
const ceo = db.agents.getCeoForCompany(company.id);
const ceoLastWake = ceo?.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0;
const ceoLastAttempt = ceo ? attemptedAtByAgent.get(ceo.id) ?? 0 : 0;
const ceoLastTouch = Math.max(ceoLastWake, ceoLastAttempt);
const bootstrapCooldown = 120_000;
if (ceo && !hasActiveRun({ agentId: ceo.id, companyId: company.id }) && (now - ceoLastTouch > bootstrapCooldown)) {
logger.info(`[heartbeat] Company ${company.name} has ${goalCount} goal(s) and no advancing tasks — bootstrapping CEO ${ceo.name}`);
tryWake(ceo.id, company.id, ceo.name, "bootstrap");
}
}
const scheduleKey = `schedule-${company.id}`;
const lastScheduleFired = scheduleLastFiredMap.get(scheduleKey) ?? 0;
if (now - lastScheduleFired >= 60_000) {
dispatchAutomation("schedule", { companyId: company.id });
scheduleLastFiredMap.set(scheduleKey, now);
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.heartbeat.company_tick_failed",
message: `Heartbeat tick failed for company ${company.id}`,
context: { companyId: company.id },
});
}
}
} catch (error) {
logAppError(logger, error, {
kind: "recoverable",
code: "scheduler.heartbeat.tick_failed",
message: "Heartbeat scheduler tick raised; next interval will retry",
});
}
}
/**
* Re-arm the interval after OS resume. On Windows (and to a lesser
* extent macOS App Nap), `setInterval` is throttled to ≥1 minute and
* sometimes stalls entirely across sleep/standby cycles — which is
* exactly why agents went 4+ days without a single successful
* heartbeat. We can't trust the handle to fire on schedule after
* resume, so we clear it, run one tick immediately to catch up, and
* restart the interval with a fresh timer. Idempotent; safe to call
* during normal operation.
*/
function rearm(trigger: string) {
if (isQuitting()) return;
logger.info(`[heartbeat] Re-arming scheduler (${trigger})`);
if (timer) clearInterval(timer);
timer = setInterval(runTick, 15_000);
runTick();
}
let resumeListener: (() => void) | null = null;
let unlockListener: (() => void) | null = null;
function start() {
if (timer) clearInterval(timer);
timer = setInterval(runTick, 15_000);
// Windows/macOS: when the OS resumes from sleep or the screen
// unlocks, setInterval timers set before sleep are unreliable
// (Windows modern-standby throttles background timers to minute
// granularity and can drop pending fires entirely). Listening to
// powerMonitor is the documented Electron path to get a
// deterministic kick that doesn't depend on a user clicking the
// window. Without this the CEO heartbeat silently stops firing
// whenever the laptop lid closes overnight.
if (!resumeListener) {
resumeListener = () => rearm("power_resume");
powerMonitor.on("resume", resumeListener);
}
if (!unlockListener) {
unlockListener = () => rearm("screen_unlock");
// `unlock-screen` fires on Windows/macOS when the user returns
// from a locked screen without a full sleep cycle — common path
// for "I left the app open, locked, came back, agents dead".
powerMonitor.on("unlock-screen", unlockListener);
}
}
function stop() {
if (timer) clearInterval(timer);
timer = null;
if (resumeListener) {
powerMonitor.removeListener("resume", resumeListener);
resumeListener = null;
}
if (unlockListener) {
powerMonitor.removeListener("unlock-screen", unlockListener);
unlockListener = null;
}
}
return { start, stop };
+39
View File
@@ -291,6 +291,43 @@ function buildConnectorCatalog(
.filter((c) => c.enabledModels.length > 0);
}
/**
* Lift the scheduler-written `heartbeatSkip` blob out of
* `runtime_state_json` into a typed PromptContext slice. The heartbeat
* scheduler writes the reader-free fields {reason, streak, since,
* lastTickAt}; we validate the shape here because `getRuntimeState`
* returns `Record<string, unknown>` and bad JSON shouldn't poison the
* prompt. Missing / malformed entries return null — the prompt section
* is omitted rather than rendered with placeholder values.
*/
function readHeartbeatSkip(
db: AppDatabase,
agentId: string,
): {
reason: string;
streak: number;
since: string;
lastTickAt: string;
} | null {
const state = db.agents.getRuntimeState(agentId);
const raw = state["heartbeatSkip"];
if (!raw || typeof raw !== "object") return null;
const record = raw as Record<string, unknown>;
const reason = record.reason;
const streak = record.streak;
const since = record.since;
const lastTickAt = record.lastTickAt;
if (
typeof reason !== "string" ||
typeof streak !== "number" ||
typeof since !== "string" ||
typeof lastTickAt !== "string"
) {
return null;
}
return { reason, streak, since, lastTickAt };
}
/**
* Assemble the full run payload from resolved entities and dependencies.
* This is the core composition step that was previously inlined in dispatchRunToWorker.
@@ -308,6 +345,7 @@ export function buildRunPayload(params: RunDispatchParams, deps: TurnContextDeps
const initialHandoff = resolveSessionHandoff(agent.id, deps);
const { taskSessionId, sessionHandoff } = resolveSession(db, agent, companyId, connector.id, initialHandoff, deps);
const skills = resolveSkills(companyId, agent.id, connector.id, agent.name, deps);
const heartbeatSkip = readHeartbeatSkip(db, agent.id);
const systemPrompt = apiUrl ? buildTaskPrompt({
agent,
@@ -353,6 +391,7 @@ export function buildRunPayload(params: RunDispatchParams, deps: TurnContextDeps
.filter((workspace) => workspace.companyId === companyId)
.map((workspace) => ({ name: workspace.name, localPath: workspace.localPath })),
companyDocuments: snapshot.documents.filter((d) => d.companyId === companyId),
heartbeatSkip,
}) : null;
const thinkingEffort = extractThinkingEffort(agent.metadataJson, deps.logger, agent.id);
+18
View File
@@ -45,6 +45,20 @@ export interface PromptContext {
companySkillCatalog?: Array<{ name: string; slug: string; description: string | null }>;
/** Company workspaces visible to the current run. */
companyWorkspaces?: Array<{ name: string; localPath: string }>;
/**
* Pending-skip context from the heartbeat scheduler. Present only
* when the scheduler most recently refused to dispatch this agent
* (e.g. worker not ready, cooldown). Surfacing it into the system
* prompt is the reader half of the scheduler-side write — without
* it, a multi-day skip streak would be invisible to the agent even
* though `runtime_state_json.heartbeatSkip` tracks it precisely.
*/
heartbeatSkip?: {
reason: string;
streak: number;
since: string;
lastTickAt: string;
} | null;
}
export function buildTaskPrompt(ctx: PromptContext): string {
@@ -69,6 +83,10 @@ export function buildTaskPrompt(ctx: PromptContext): string {
parts.push(`\n${m.wakeReasonHeading}\n${explanation}\n`);
}
if (ctx.heartbeatSkip) {
parts.push(`\n${m.heartbeatSkipSection(ctx.heartbeatSkip)}`);
}
parts.push(`\n${m.timeAwarenessSection}`);
// Taskless runs used to emit a "respond with a brief status summary"
@@ -271,6 +271,23 @@ export function importantRulesSection(companyId: string, agentId: string, runId:
export const wakeReasonHeading = "# Wake Reason";
export function heartbeatSkipSection(args: {
reason: string;
streak: number;
since: string;
lastTickAt: string;
}): string {
return `# Heartbeat Skip Status
The scheduler most recently declined to dispatch this agent.
- Reason: ${args.reason}
- Consecutive skips: ${args.streak}
- First skip in this streak: ${args.since}
- Last tick: ${args.lastTickAt}
If this streak is long and unexplained, it likely means a subsystem (worker pool, connector, quota, checkpoint hold) has been blocking wakes. Check the subsystem before assuming silence means nothing is wrong.
`;
}
export const timeAwarenessSection = `# Time Awareness
Consider what phase of operations the company is in and prioritize accordingly:
- Early in a sprint cycle → planning, task breakdown, delegation