mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: clarify code queue split-brain liveness

Browse Source
This commit is contained in:
Codex
2026-05-20 01:41:16 +00:00
parent b01907739c
commit ada6da3da6
11 changed files with 264 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/code-queue-liveness-fixtures.ts
+8
View File
@@ -179,6 +179,7 @@ function checkStaleActiveOwnerExpired(): FixtureCheck {
const diagnostics = schedulerDiagnostics([task], []);
assertCondition(decision.allowed === true && decision.reason === "owner-heartbeat-expired", "expired owner heartbeat should be the stale recovery gate", { decision });
assertCondition(diagnostics.state === "stale-active", "diagnostics must mark stale active only after owner heartbeat expiry", diagnostics as unknown as Record<string, unknown>);
assertCondition(diagnostics.effectiveLiveness === "at-risk" && diagnostics.recommendedAction === "investigate-heartbeat-risk", "expired heartbeat diagnostics must surface at-risk liveness", diagnostics as unknown as Record<string, unknown>);
assertCondition(diagnostics.staleRecoveryCandidateTaskIds.includes(task.id), "expired owner heartbeat should create a stale candidate", diagnostics as unknown as Record<string, unknown>);
return {
name: "code-queue:stale-active-owner-expired",
@@ -186,6 +187,8 @@ function checkStaleActiveOwnerExpired(): FixtureCheck {
detail: {
decision,
state: diagnostics.state,
effectiveLiveness: diagnostics.effectiveLiveness,
recommendedAction: diagnostics.recommendedAction,
staleRecoveryCandidateTaskIds: diagnostics.staleRecoveryCandidateTaskIds,
},
};
@@ -203,6 +206,8 @@ function checkControlPlaneSplitBrainDiagnostics(): FixtureCheck {
oaPublisher: null,
});
assertCondition(diagnostics.state === "split-brain" && diagnostics.splitBrain === true, "master postgres-control-plane must report split-brain when DB active has fresh scheduler heartbeat", diagnostics as unknown as Record<string, unknown>);
assertCondition(diagnostics.splitBrainLive === true && diagnostics.effectiveLiveness === "live", "fresh split-brain diagnostics must be degraded but live", diagnostics as unknown as Record<string, unknown>);
assertCondition(diagnostics.recommendedAction === "continue-supervision", "fresh split-brain diagnostics must recommend continued supervision", diagnostics as unknown as Record<string, unknown>);
assertCondition(diagnostics.schedulerActiveRunSlotCount === 0 && diagnostics.databaseActiveTaskCount === 1, "split-brain fixture should preserve the exact control-plane divergence", diagnostics as unknown as Record<string, unknown>);
return {
name: "code-queue:control-plane-split-brain-diagnostics",
@@ -210,6 +215,9 @@ function checkControlPlaneSplitBrainDiagnostics(): FixtureCheck {
detail: {
state: diagnostics.state,
splitBrain: diagnostics.splitBrain,
splitBrainLive: diagnostics.splitBrainLive,
effectiveLiveness: diagnostics.effectiveLiveness,
recommendedAction: diagnostics.recommendedAction,
executionStateSource: diagnostics.executionStateSource,
databaseActiveTaskIds: diagnostics.databaseActiveTaskIds,
schedulerActiveRunSlotCount: diagnostics.schedulerActiveRunSlotCount,
scripts/src/code-queue.ts
+52 -1
View File
@@ -239,7 +239,14 @@ function upstreamError(response: unknown): string {
if (typeof bodyError === "string") {
const requestId = typeof body?.requestId === "string" ? ` requestId=${body.requestId}` : "";
const providerId = typeof body?.providerId === "string" ? ` providerId=${body.providerId}` : "";
return `${bodyError}${providerId}${requestId}`;
const observationHint = /\b(microservice|proxy|provider|tunnel|k3sctl|adapter|backend-core|offline|unavailable|timed out|timeout)\b/iu.test(bodyError)
? " The stable code-queue proxy failure is observation-path evidence only; from the supervisor or main-server CLI environment, also check `bun scripts/cli.ts codex queues`, `codex tasks`, or `codex task <taskId>` before declaring the work unevaluable or stopped."
: "";
return `${bodyError}${providerId}${requestId}${observationHint}`;
}
if (typeof record.codeQueueObservationNote === "string") {
const commands = Array.isArray(record.recommendedCommands) ? ` recommendedCommands=${JSON.stringify(record.recommendedCommands)}` : "";
return `${record.codeQueueObservationNote}${commands}`;
}
const status = typeof record.status === "number" ? `HTTP ${record.status}` : "upstream request failed";
return `${status}: ${JSON.stringify(response).slice(0, 1200)}`;
@@ -335,13 +342,56 @@ function compactSchedulerHeartbeat(value: unknown): Record<string, unknown> | nu
};
}
function splitBrainLiveFromDiagnostics(record: Record<string, unknown>): boolean {
if (typeof record.splitBrainLive === "boolean") return record.splitBrainLive;
const state = String(record.state ?? record.health ?? "").toLowerCase();
const riskTaskIds = Array.from(new Set([
...stringList(record.heartbeatRiskTaskIds),
...stringList(record.heartbeatExpiredTaskIds),
...stringList(record.heartbeatMissingTaskIds),
...stringList(record.staleRecoveryCandidateTaskIds),
]));
return state === "split-brain" && stringList(record.heartbeatFreshTaskIds).length > 0 && riskTaskIds.length === 0;
}
function effectiveLivenessFromDiagnostics(record: Record<string, unknown>): string {
if (typeof record.effectiveLiveness === "string" && record.effectiveLiveness.length > 0) return record.effectiveLiveness;
if (stringList(record.heartbeatRiskTaskIds).length > 0
|| stringList(record.heartbeatExpiredTaskIds).length > 0
|| stringList(record.heartbeatMissingTaskIds).length > 0
|| stringList(record.staleRecoveryCandidateTaskIds).length > 0) {
return "at-risk";
}
if (splitBrainLiveFromDiagnostics(record)) return "live";
return String(record.state ?? record.health ?? "unknown") === "healthy" ? "healthy" : "degraded";
}
function recommendedActionFromDiagnostics(record: Record<string, unknown>): string {
if (typeof record.recommendedAction === "string" && record.recommendedAction.length > 0) return record.recommendedAction;
const effectiveLiveness = effectiveLivenessFromDiagnostics(record);
if (effectiveLiveness === "at-risk") return "investigate-heartbeat-risk";
if (effectiveLiveness === "live") return "continue-supervision";
if (effectiveLiveness === "degraded") return "observe-degraded";
return "none";
}
function compactExecutionDiagnostics(value: unknown): Record<string, unknown> | null {
const record = asRecord(value);
if (record === null) return null;
const heartbeatRiskTaskIds = Array.from(new Set([
...stringList(record.heartbeatRiskTaskIds),
...stringList(record.heartbeatExpiredTaskIds),
...stringList(record.heartbeatMissingTaskIds),
...stringList(record.staleRecoveryCandidateTaskIds),
])).sort();
return {
state: record.state ?? record.health ?? null,
degraded: record.degraded ?? null,
splitBrain: record.splitBrain ?? null,
splitBrainLive: splitBrainLiveFromDiagnostics(record),
effectiveLiveness: effectiveLivenessFromDiagnostics({ ...record, heartbeatRiskTaskIds }),
recommendedAction: recommendedActionFromDiagnostics({ ...record, heartbeatRiskTaskIds }),
livenessSummary: record.livenessSummary ?? null,
executionStateSource: record.executionStateSource ?? null,
controlPlane: record.controlPlane ?? null,
databaseActiveTaskCount: record.databaseActiveTaskCount ?? null,
@@ -353,6 +403,7 @@ function compactExecutionDiagnostics(value: unknown): Record<string, unknown> |
heartbeatExpiredTaskIds: record.heartbeatExpiredTaskIds ?? [],
heartbeatMissingTaskIds: record.heartbeatMissingTaskIds ?? [],
staleRecoveryCandidateTaskIds: record.staleRecoveryCandidateTaskIds ?? [],
heartbeatRiskTaskIds,
traceGapTaskIds: record.traceGapTaskIds ?? [],
traceGapNotStaleTaskIds: record.traceGapNotStaleTaskIds ?? [],
lastSchedulerHeartbeatAt: record.lastSchedulerHeartbeatAt ?? null,
scripts/src/microservices.ts
+15 -1
View File
@@ -32,7 +32,21 @@ export function coreInternalFetch(path: string, init?: { method?: string; body?:
const command = dockerCoreFetchCommand(path, init);
const result = runCommand(command, repoRoot);
if (result.exitCode !== 0) {
return { ok: false, exitCode: result.exitCode, stdoutTail: result.stdout.slice(-1200), stderrTail: result.stderr.slice(-1200) };
const codeQueueStableProxy = path.startsWith("/api/microservices/code-queue/");
return {
ok: false,
exitCode: result.exitCode,
stdoutTail: result.stdout.slice(-1200),
stderrTail: result.stderr.slice(-1200),
...(codeQueueStableProxy ? {
codeQueueObservationNote: "The stable /api/microservices/code-queue path could not be reached from this CLI container. This does not by itself mean Code Queue tasks are unevaluable or stopped. From the supervisor or main-server CLI environment, use the existing codex queues/tasks/task observation commands instead of treating this container-local proxy failure as liveness evidence.",
recommendedCommands: [
"bun scripts/cli.ts codex queues",
"bun scripts/cli.ts codex tasks --limit 20",
"bun scripts/cli.ts codex task <taskId>",
],
} : {}),
};
}
try {
return JSON.parse(result.stdout.trim()) as unknown;