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
116 changes: 99 additions & 17 deletions .pi/extensions/fm-primary-pi-watch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,8 @@ const armScript = `${fmRoot}/bin/fm-watch-arm.sh`;
const marker = `${state}/.pi-watch-extension-loaded`;
const handoffDir = `${state}/extensions/pi-primary-watch`;
const actionableHandoff = `${handoffDir}/session-replacement-actionable.json`;
const extensionLog = `${state}/.watch-extension.log`;
const extensionLogMaxLines = extensionLogKeepLines();
const extensionVersion = `sha256:${createHash("sha256").update(readFileSync(extensionFile)).digest("hex")}`;
const retryBaseMs = positiveInteger("FM_WATCH_REARM_RETRY_BASE_MS", 250);
const retryMaxMs = positiveInteger("FM_WATCH_REARM_RETRY_MAX_MS", 4000);
Expand Down Expand Up @@ -213,6 +215,18 @@ function positiveInteger(name: string, fallback: number): number {
return Math.floor(value);
}

// Opt-in bound for the extension diagnostic log: only a positive
// FM_WATCH_EXTENSION_LOG_KEEP_LINES enables logging, so the default run
// writes nothing. Unset, empty, non-numeric, zero, and negative values
// disable the log entirely instead of falling back to a silent default.
function extensionLogKeepLines(): number {
const raw = process.env.FM_WATCH_EXTENSION_LOG_KEEP_LINES;
if (raw === undefined || raw.trim() === "") return 0;
const value = Math.floor(Number(raw));
if (!Number.isFinite(value) || value <= 0) return 0;
return value;
}

function parentPid(pid: string): string {
const result = spawnSync("ps", ["-o", "ppid=", "-p", pid], { encoding: "utf8" });
if (result.status !== 0) return "";
Expand All @@ -228,6 +242,21 @@ function pidAlive(pid: string): boolean {
}
}

// An arm child whose process is gone but whose close event has not fired yet
// (stdio pipes still held) must not keep the single-flight slot: neither a
// repair call nor a scheduled retry would start anything until that close
// finally fires. Callers that gate on slot occupancy use this instead of
// owner.child so both paths can always recover.

function liveArmChild(owner: SessionGeneration): ChildProcess | null {
const child = owner.child;
if (!child) return null;
if (child.exitCode !== null || child.signalCode !== null) return null;
const pid = child.pid;
if (pid === undefined || !pidAlive(String(pid))) return null;
return child;
}

function lockOwnership(): LockOwnership {
let lockPid = "";
try {
Expand Down Expand Up @@ -314,6 +343,35 @@ function nodeErrorCode(error: unknown): string {
: "";
}

// Bounded diagnostic record for restore attempts, readiness timeouts, and
// handling-confirmation targets and results. Opt-in through
// FM_WATCH_EXTENSION_LOG_KEEP_LINES and off by default: a disabled log
// returns before touching the filesystem, so it never creates its file.
// Purely observational: a logging failure never changes supervision
// behavior. docs/watcher-continuity.md owns what the arm layer already
// records; this file is the extension side.
function appendExtensionLog(detail: string): void {
if (extensionLogMaxLines <= 0) return;
try {
mkdirSync(state, { recursive: true });
const cleaned = detail.replace(/[\r\n\t]+/g, " ").slice(0, 512);
const line = `${new Date().toISOString()} pid=${process.pid} ${cleaned}`;
let previous = "";
try {
previous = readFileSync(extensionLog, "utf8");
} catch (error) {
if (nodeErrorCode(error) !== "ENOENT") return;
}
const joined = `${previous}${previous === "" || previous.endsWith("\n") ? "" : "\n"}${line}\n`;
const kept = joined.split("\n").slice(-(extensionLogMaxLines + 1)).join("\n");
const temporary = `${extensionLog}.tmp-${process.pid}`;
writeFileSync(temporary, kept, { mode: 0o600 });
renameSync(temporary, extensionLog);
} catch {
// Diagnostic only: never fail supervision for observability.
}
}

function createPendingActionable(message: string, predecessorArmPid: string): PendingActionableClose {
return {
version: 1,
Expand Down Expand Up @@ -611,6 +669,7 @@ export default function (pi: ExtensionAPI) {
function confirmHandlingDelivery(recovery: { generation: string; watcherPid: string }): {
ok: boolean;
detail: string;
superseded?: boolean;
} {
try {
const result = spawnSync(
Expand All @@ -623,6 +682,11 @@ export default function (pi: ExtensionAPI) {
},
);
if (result.status === 0) return { ok: true, detail: "" };
if (result.status === 3) {
// The marker advanced past this restoration's generation mid-restore,
// so a newer pipeline owns the episode now: superseded, not rejected.
return { ok: false, detail: "", superseded: true };
}
const stderr = (result.stderr || "").trim();
return {
ok: false,
Expand All @@ -638,16 +702,15 @@ export default function (pi: ExtensionAPI) {
}

function confirmHandlingDeliveryWithRetry(
owner: SessionGeneration,
recovery: { generation: string; watcherPid: string },
): { ok: boolean; detail: string } {
const snapshot = (): { generation: string; watcherPid: string } => {
const current = owner.child ? armRecovery.get(owner.child) : undefined;
return current ?? recovery;
};
const first = confirmHandlingDelivery(snapshot());
if (first.ok) return first;
return confirmHandlingDelivery(snapshot());
): { ok: boolean; detail: string; superseded?: boolean } {
// Confirm the restoration's own recovery token, never a fresh snapshot of
// the current arm child: a successor replaced during the restore window
// must not turn this delivery into a false rejection, and a retry must
// not retire a newer healthy watcher.
const first = confirmHandlingDelivery(recovery);
if (first.ok || first.superseded) return first;
return confirmHandlingDelivery(recovery);
}

function offerWakeToBranch(message: string): Promise<void> | null {
Expand All @@ -668,11 +731,25 @@ export default function (pi: ExtensionAPI) {
): Promise<boolean> {
if (!generationIsLive(owner)) return false;
if (recovery) {
const confirmed = confirmHandlingDeliveryWithRetry(owner, recovery);
if (!confirmed.ok) {
const watcherPid = recovery.watcherPid;
if (!pidAlive(watcherPid)) {
await retireArm(owner.child);
const confirmed = confirmHandlingDeliveryWithRetry(recovery);
appendExtensionLog(
`confirm generation=${recovery.generation} watcherPid=${recovery.watcherPid} result=${confirmed.ok ? "confirmed" : confirmed.superseded ? "superseded" : "rejected"}`,
);
// A superseded result means a newer pipeline owns this episode now: it
// routes like a confirmed delivery below, with no failure appended, and
// retires nothing.
if (!confirmed.ok && !confirmed.superseded) {
const failedPid = recovery.watcherPid;
const current = owner.child;
const currentRecovery = current ? armRecovery.get(current) : undefined;
if (
current &&
currentRecovery?.watcherPid === failedPid &&
currentRecovery?.generation === recovery.generation &&
!pidAlive(failedPid)
) {
appendExtensionLog(`retire pid=${failedPid} reason=confirm-failure`);
await retireArm(current);
}
return await sendWake(owner, `${message}\n\n${confirmed.detail}`, pending);
}
Expand Down Expand Up @@ -850,7 +927,7 @@ export default function (pi: ExtensionAPI) {
// been idle.
const deferred = owner.deferredClose;
owner.deferredClose = null;
if (deferred && !owner.child && !owner.retryTimer) {
if (deferred && !liveArmChild(owner) && !owner.retryTimer) {
scheduleRetry(owner, deferred.message, deferred.predecessorArmPid);
}
}
Expand Down Expand Up @@ -912,10 +989,14 @@ export default function (pi: ExtensionAPI) {
if (!generationIsLive(owner)) return { failure: "" };
const replacement = startArm(owner, predecessorArmPid);
const successorChild = owner.child;
appendExtensionLog(
`restore attempt=${attempt} predecessor=${predecessorArmPid || "none"} start=${replacement.ok ? `ok pid=${successorChild?.pid ?? "none"}` : "failed"}`,
);
if (replacement.ok && successorChild && await waitForReadiness(successorChild)) {
return { failure: "", recovery: armRecovery.get(successorChild) };
}
if (replacement.ok) {
appendExtensionLog(`restore attempt=${attempt} readiness=timeout pid=${successorChild?.pid ?? "none"}`);
failure = "watcher: FAILED - Pi extension could not verify a ready successor watcher";
if (!(await retireArm(successorChild))) {
return {
Expand All @@ -931,11 +1012,12 @@ export default function (pi: ExtensionAPI) {
if (attempt === retryLimit) break;
await waitForRetry(attempt + 1);
}
appendExtensionLog(`restore exhausted attempts=${retryLimit + 1} outcome=hand-to-main`);
return { failure: `${failure}\nwatcher: FAILED - Pi extension could not restore watcher continuity after ${retryLimit} retries` };
}

function scheduleRetry(owner: SessionGeneration, message: string, predecessorArmPid: string): void {
if (!generationIsLive(owner) || owner.child || owner.retryTimer) return;
if (!generationIsLive(owner) || liveArmChild(owner) || owner.retryTimer) return;
const ownership = lockOwnership();
if (ownership !== "owned") {
surfaceFailure(owner, `watcher: FAILED - Pi extension cannot restore continuity because this session no longer owns the lock\n${message}`);
Expand Down Expand Up @@ -969,7 +1051,7 @@ export default function (pi: ExtensionAPI) {
};
}
publishGenerationOwner(owner, "active");
if (owner.child) {
if (liveArmChild(owner)) {
return {
ok: true,
message: `watcher: unchanged - Pi extension already owns an arm child; no manual re-arm needed; ${repairOnlyHint}`,
Expand Down
11 changes: 11 additions & 0 deletions bin/fm-wake-lib.sh
Original file line number Diff line number Diff line change
Expand Up @@ -813,6 +813,17 @@ _fm_recovery_marker_begin_handling() {
fi
case "$line" in
pending:handling:*|announced:handling:*) ;;
acked:handling:*|acked:downtime:*)
# An already-retired episode confirms as a no-op when the caller names
# its generation: the drain acknowledged it after the successor started
# but before the delivery confirmation ran. Without a named generation
# there is nothing to match, so keep the rejection.
# docs/watcher-continuity.md owns the recovery-episode contract.
if [ -z "$expected_generation" ]; then
fm_lock_release "$lock"
return 1
fi
;;
pending:downtime:*)
if ! _fm_recovery_marker_write_locked "$marker" handling "$generation"; then
fm_lock_release "$lock"
Expand Down
1 change: 1 addition & 0 deletions docs/configuration.md
Original file line number Diff line number Diff line change
Expand Up @@ -2351,6 +2351,7 @@ FM_WATCH_REARM_RETRY_MAX_MS=4000 # Pi/OpenCode adapter cap for exponential con
FM_WATCH_REARM_RETRY_LIMIT=5 # Pi/OpenCode adapter launch-failure retries before surfacing restoration failure
FM_WATCH_CYCLE_LOG_MAX_BYTES=262144 # size cap for the arm-owned watcher lifecycle ledger
FM_WATCH_CYCLE_LOG_KEEP_LINES=1000 # newest complete lifecycle rows considered when the ledger is capped
FM_WATCH_EXTENSION_LOG_KEEP_LINES=0 # opt-in Pi extension diagnostic log (state/.watch-extension.log); unset, empty, non-numeric, zero, or negative disables logging, a positive value keeps that many newest rows; logging never changes supervision behavior
FM_WATCHER_STALE_GRACE=300 # defaults to FM_GUARD_GRACE if set, else the poll-derived grace (docs/turnend-guard.md "Guard grace and the poll cadence"); seconds before a fresh arm refuses a live holder's stale beacon (attached arms: FM_WATCHER_STALL_BOUND)
FM_WATCHER_STALL_BOUND= # live-holder stall bound; default and arm/re-arm behavior: docs/turnend-guard.md "Guard grace and the poll cadence"
FM_SIGNAL_GRACE=30 # seconds to coalesce nearby status and turn-end signals into one wake
Expand Down
17 changes: 14 additions & 3 deletions docs/watcher-continuity.md
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ Each adapter:
- Preserves one child or scheduled retry at a time.
- Applies bounded exponential retry after an unexpected or failed close.

Pi treats an arm child whose process is already gone as an empty slot even while its close event is still pending, so a repair call or a scheduled retry starts a fresh arm instead of answering unchanged.
A failed follow-up never cancels continuity restoration.

### Pi session replacement
Expand Down Expand Up @@ -133,14 +134,19 @@ After an actionable Pi, omp, or OpenCode child close, the adapter:

1. Waits for the predecessor process to close.
2. Starts and verifies one singleton successor.
3. Confirms the handling handoff against that successor before scheduling the follow-up.
3. Confirms the handling handoff before scheduling the follow-up: Pi confirms against the restoration's own recovery token, while omp and OpenCode confirm against the current successor.
4. Delivers the original wake.

A complete Pi reason line can be observed while the predecessor is still finishing durable cleanup.
That line is retained for replacement handoff, but the adapter never treats that already-ready predecessor as its own successor.

If the handoff confirmation fails, the adapter retries it once against the current generation and successor.
A failed confirmation is a restoration failure: the adapter classifies the error, retires a successor that is no longer alive, and surfaces exactly one typed message.
If the handoff confirmation fails, the adapter retries it once: Pi against that same token, omp and OpenCode against the current generation and successor.
A failed confirmation is a restoration failure: the adapter classifies the error and surfaces exactly one typed message.
Pi retires the current successor only when the failed token names its exact watcher pid and generation and that pid is no longer alive, while omp and OpenCode retire the current successor whenever the restoration's watcher pid is no longer alive.
On Pi a generation mismatch means a newer pipeline superseded this delivery mid-restore, so the wake routes like a confirmed delivery, with no failure appendix, and nothing is retired.
An already-acknowledged episode confirms as a no-op when the confirmation names its generation, because the drain acknowledged it after the successor started but before the confirmation ran.
The Pi extension diagnostic log is opt-in and off by default: only a positive FM_WATCH_EXTENSION_LOG_KEEP_LINES value appends restore attempts, readiness timeouts, and confirmation targets and results to state/.watch-extension.log, a bounded record that never changes supervision behavior.
docs/configuration.md owns the knob's default and accepted values.
A failed confirmation is never swallowed.

### Readiness timeout and retry
Expand Down Expand Up @@ -425,6 +431,9 @@ The same suite covers ordinary same-process session replacement for `/new`, `/re
- Repeated transitions with exactly one live cycle.
- Disappearance of the shutting-down refusal after a valid replacement activates.
- Terminal quit still refusing late rearm.
- A mid-restore marker advance that delivers the wake with no rejection appendix, offers it to an accepting supervision branch like a confirmed delivery, and records the attempt and the confirm result in the bounded extension log when opted in.
- A failed confirmation for a stale successor that spares a newer arm started by a repair.
- A repair, a scheduled retry, and a deferred close over a dead-but-unclosed arm child that each start a fresh arm instead of stalling.

The guard and session-start suites prove that active generation evidence tolerates a fresh-beacon handoff.
They also prove that a legacy or handoff-phase watcher marker from an absent replacement extension still raises the outage diagnostic.
Expand All @@ -444,6 +453,8 @@ They also prove that a legacy or handoff-phase watcher marker from an absent rep
- A watcher close inside the handling window that must leave the printed acknowledgement valid.
- A re-arm whose recovery cycle is slowed after confirmation and must still surface rather than read as a watcher that stayed live.
- The self-healing moved-generation acknowledgement that consumes its handled rows and names its remedy.
- The already-acknowledged confirmation no-op for a matching generation, with its mismatched-generation, dead-pid, and lock-mismatch rejections preserved.
- The manual-restart generation churn that makes a confirmation for the churned generation report a mismatch, which an arm check without a reopen leaves in place.
- A take-over that stays quiet after a confirmed TERM, still surfaces queued work and self-exit downtime, and attaches without stopping a cycle the named arm does not own.
- The disposable-checkout arm refusal.
- The home-gone and state-gone watcher exits.
Expand Down
Loading
Loading