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
5 changes: 3 additions & 2 deletions src/harness/bridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -263,10 +263,11 @@ export interface InstanceUpdate {
changes: Record<UpdateClass, DefinitionChange[]>;
}

/** Pushed main → renderer for the attached environment: each daemon event in seq order, or a resync snapshot. */
/** Pushed main → renderer: attach metadata, each daemon event in seq order, or a resync snapshot. */
export type DaemonEventPayload =
| { envId: string; seq: number; at: number; ev: DaemonEvent }
| { envId: string; snapshot: Snapshot };
| { envId: string; snapshot: Snapshot }
| { envId: string; welcome: { daemon: Snapshot['daemon']; head: number } };

/** One GitHub App installation the signed-in user can reach. */
export interface GithubInstallation {
Expand Down
8 changes: 4 additions & 4 deletions src/main/instances/daemon-client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -56,11 +56,11 @@ export interface DaemonClientDeps {
since(): number | null;
saveSeq(seq: number): void;
onEvent(seq: number, at: number, ev: DaemonEvent): void;
/** A resync: the snapshot replaces everything the owner projected. */
/** A resync: the owner applies this snapshot in place of missed events. */
onSnapshot(snapshot: Snapshot): void;
onState(state: AttachState, detail: string): void;
/** Each welcome (the daemon's version, for update checks). */
onWelcome?(daemon: { version: string; build: string; protocol: number }): void;
/** Each welcome: the current build and the head through which events are replayed. */
onWelcome?(daemon: Snapshot['daemon'], head: number): void;
client: { app: string; build: string };
now?(): number;
/** Test seam. */
Expand Down Expand Up @@ -203,7 +203,7 @@ export class DaemonClient {
}
this.failures = 0;
this.everAttached = true;
this.deps.onWelcome?.({ ...frame.daemon, protocol: frame.protocol });
this.deps.onWelcome?.({ ...frame.daemon, protocol: frame.protocol }, frame.head);
this.setState('attached');
if (frame.replay === 'resync') {
held = [];
Expand Down
5 changes: 4 additions & 1 deletion src/main/instances/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -268,8 +268,11 @@ function attach(envId: string): void {
emitDaemon({ envId, snapshot });
changed(envId);
},
onWelcome: (daemon) => {
onWelcome: (daemon, head) => {
const previous = daemonBuilds.get(envId);
daemonBuilds.set(envId, daemon.build);
emitDaemon({ envId, welcome: { daemon, head } });
if (previous && previous !== daemon.build) log.info('instance.daemon-version-changed', { envId, previous, build: daemon.build, version: daemon.version });
changed(envId);
},
onState: (state: AttachState, detail) => {
Expand Down
131 changes: 124 additions & 7 deletions src/renderer/instance-store.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,13 @@
* is a replay overlap and is skipped. Events that arrive before the first
* snapshot, or ahead of a missing seq, wait in a buffer; a gap that the
* buffer cannot close asks the owner for a resync (once per cursor).
* - A snapshot replaces everything and moves the cursor to its head;
* buffered events after the head then apply on top.
* - A snapshot replaces the projection and moves the cursor to its head;
* buffered events after the head then apply on top. The same daemon
* keeps an in-progress or failed restart, and recovers a drain whose
* original turns are still running; a different version or build clears
* the update. A welcome updates the version and build without replacing
* the projection. The same build recovers that drain the same way; a
* different version or build clears the update.
* - Kept: items, backlog order, sessions, capacity, instance status, GitHub
* state, open questions, and live turn buffers (the recorded dialect of
* every turn still running, so a thread opened mid-turn and the board's
Expand Down Expand Up @@ -38,6 +43,8 @@ export type StoreChange =
| { kind: 'instances' }
| { kind: 'reset'; envId: string | null }
| { kind: 'snapshot'; envId: string }
/** Attach metadata or the daemon update deadline changed. */
| { kind: 'daemon'; envId: string }
/** An optimistic local reorder, before the daemon confirms it. */
| { kind: 'order'; envId: string }
| { kind: 'event'; envId: string; seq: number; ev: DaemonEvent };
Expand All @@ -50,18 +57,26 @@ export interface EnvDaemonState {
github: GithubAuth;
orchestratorSessionId: string | null;
capacity: Capacity;
/** Set by `daemon.upgrading` until the next snapshot. */
/** Set by `daemon.upgrading` until a changed build returns or the restart times out. */
upgrading: 'drain' | 'now' | null;
upgradeError: string | null;
/** The definition's repositories, from the snapshot or the latest `instance.definition`. */
repos: { github: string; dir: string }[];
}

export interface InstanceStoreOptions {
/** The buffer cannot close a gap in seq: fetch a snapshot (the owner calls applySnapshot). */
requestResync(envId: string): void;
/** Restart deadline. An attached drain's original turns do not count; staying disconnected still bounds the wait. Test seam. */
upgradeTimeoutMs?: number;
}

const EMPTY_CAPACITY: Capacity = { agents: {}, workers: { running: 0, max: 0 }, paused: false };
const UPGRADE_TIMEOUT_MS = 120_000;

function sameDaemon(a: Snapshot['daemon'], b: Snapshot['daemon']): boolean {
return a.build === b.build && a.version === b.version;
}

/** Consecutive text deltas with the same parentId merge, and `thinking` is dropped. */
export function recordLive(events: HarnessEvent[], event: HarnessEvent): void {
Expand Down Expand Up @@ -108,11 +123,75 @@ export function createInstanceStore(opts: InstanceStoreOptions) {
const inflight = new Map<string, InflightTurn>();
/** The latest tool summary per session while its turn runs. */
const lastTool = new Map<string, string>();
let upgradeTimer: ReturnType<typeof setTimeout> | null = null;
/** Keep the phase after a timeout so a same-build snapshot or welcome can recover a still-running drain. */
let upgradePhase: 'draining' | 'restarting' | null = null;
/** Only the turns running when the upgrade began can delay its restart deadline. */
const drainTurns = new Set<string>();
/** An old upgrading event replayed through this head belongs to the build we replaced. */
let completedUpgradeThrough: number | null = null;

function emit(change: StoreChange): void {
for (const cb of listeners) cb(change);
}

function cancelUpgradeTimer(): void {
if (upgradeTimer) clearTimeout(upgradeTimer);
upgradeTimer = null;
}

function finishUpgrade(): void {
cancelUpgradeTimer();
upgradePhase = null;
drainTurns.clear();
if (!state) return;
state.upgrading = null;
state.upgradeError = null;
}

function attachedDrain(): boolean {
return !!envId
&& !!state
&& state.upgrading === 'drain'
&& upgradePhase === 'draining'
&& state.instance.status !== 'stopping'
&& instances.get(envId)?.attach === 'attached';
}

function completeDrain(): void {
if (upgradePhase !== 'draining') return;
// Replace a temporary disconnect deadline with a full restart deadline once.
cancelUpgradeTimer();
upgradePhase = 'restarting';
drainTurns.clear();
}

function recoverRunningDrain(): void {
if (!state) return;
for (const id of drainTurns) if (!inflight.has(id)) drainTurns.delete(id);
if (upgradePhase !== 'draining' || drainTurns.size === 0 || state.instance.status === 'stopping') return;
cancelUpgradeTimer();
state.upgrading = 'drain';
state.upgradeError = null;
}

function watchUpgrade(restarting = false): void {
if (!state) return;
if (drainTurns.size === 0 || (state.instance.status === 'stopping' && state.instance.detail === 'upgrading')) completeDrain();
if (!state.upgrading) return;
const waiting = upgradePhase === 'draining' && !restarting && state.instance.status !== 'stopping';
if (waiting) return;
if (upgradeTimer) return;
upgradeTimer = setTimeout(() => {
upgradeTimer = null;
if (!state || !envId || attachedDrain()) return;
state.upgrading = null;
state.upgradeError = 'The daemon update did not complete within two minutes.';
emit({ kind: 'daemon', envId });
}, opts.upgradeTimeoutMs ?? UPGRADE_TIMEOUT_MS);
upgradeTimer.unref?.();
}

function clearDaemon(): void {
state = null;
items.clear();
Expand All @@ -123,7 +202,7 @@ export function createInstanceStore(opts: InstanceStoreOptions) {
lastTool.clear();
}

function applyOne(ev: DaemonEvent): void {
function applyOne(seq: number, ev: DaemonEvent): void {
if (!state) return;
switch (ev.kind) {
case 'instance.status':
Expand Down Expand Up @@ -163,11 +242,13 @@ export function createInstanceStore(opts: InstanceStoreOptions) {
}
break;
}
case 'turn.end':
case 'turn.end': {
inflight.delete(ev.turnId);
drainTurns.delete(ev.turnId);
lastTool.delete(ev.sessionId);
for (const [askId, ask] of asks) if (ask.turnId === ev.turnId) asks.delete(askId);
break;
}
case 'ask.routed': {
const ask = asks.get(ev.askId);
if (ask) asks.set(ev.askId, { ...ask, routedTo: ev.to });
Expand All @@ -191,12 +272,19 @@ export function createInstanceStore(opts: InstanceStoreOptions) {
state.capacity = { agents: ev.agents, workers: ev.workers, paused: ev.paused };
break;
case 'daemon.upgrading':
if (completedUpgradeThrough !== null && seq <= completedUpgradeThrough) break;
cancelUpgradeTimer();
upgradePhase = ev.mode === 'drain' ? 'draining' : 'restarting';
drainTurns.clear();
if (ev.mode === 'drain') for (const id of inflight.keys()) drainTurns.add(id);
state.upgrading = ev.mode;
state.upgradeError = null;
break;
case 'turn.user':
case 'turn.notice':
break;
}
watchUpgrade();
}

function drain(): void {
Expand All @@ -206,7 +294,7 @@ export function createInstanceStore(opts: InstanceStoreOptions) {
const ev = buffered.get(seq) as DaemonEvent;
buffered.delete(seq);
cursor = seq;
applyOne(ev);
applyOne(seq, ev);
emit({ kind: 'event', envId, seq, ev });
}
for (const seq of [...buffered.keys()]) if (seq <= cursor) buffered.delete(seq);
Expand Down Expand Up @@ -234,6 +322,7 @@ export function createInstanceStore(opts: InstanceStoreOptions) {
},
upsertInstance(info: InstanceInfo): void {
instances.set(info.id, info);
if (info.id === envId) watchUpgrade(info.attach !== 'attached');
emit({ kind: 'instances' });
},
removeInstance(id: string): void {
Expand All @@ -251,6 +340,10 @@ export function createInstanceStore(opts: InstanceStoreOptions) {

/** Switch to another environment (or none): its daemon state starts empty until a snapshot. */
reset(next: string | null): void {
cancelUpgradeTimer();
upgradePhase = null;
drainTurns.clear();
completedUpgradeThrough = null;
envId = next;
cursor = null;
resyncAsked = undefined;
Expand All @@ -264,6 +357,13 @@ export function createInstanceStore(opts: InstanceStoreOptions) {

applySnapshot(snapshot: Snapshot, forEnv: string): void {
if (forEnv !== envId) return;
// A snapshot of the old process during drain cannot complete its upgrade.
const previous = state;
const unchanged = previous && sameDaemon(previous.daemon, snapshot.daemon);
if (!unchanged) {
finishUpgrade();
completedUpgradeThrough = snapshot.head;
}
clearDaemon();
state = {
envId: forEnv,
Expand All @@ -273,7 +373,8 @@ export function createInstanceStore(opts: InstanceStoreOptions) {
github: { ...snapshot.github },
orchestratorSessionId: snapshot.orchestratorSessionId,
capacity: snapshot.capacity ?? EMPTY_CAPACITY,
upgrading: null,
upgrading: unchanged ? previous.upgrading : null,
upgradeError: unchanged ? previous.upgradeError : null,
repos: snapshot.repos ?? [],
};
for (const item of snapshot.items) items.set(item.id, item);
Expand All @@ -292,12 +393,28 @@ export function createInstanceStore(opts: InstanceStoreOptions) {
}
}
}
recoverRunningDrain();
cursor = snapshot.head;
resyncAsked = undefined;
watchUpgrade();
emit({ kind: 'snapshot', envId: forEnv });
drain();
},

/** Welcome arrives on every attach, including replays that need no snapshot. */
applyWelcome(daemon: Snapshot['daemon'], head: number, forEnv: string): void {
if (forEnv !== envId || !state) return;
if (!sameDaemon(state.daemon, daemon)) {
completedUpgradeThrough = head;
finishUpgrade();
} else {
recoverRunningDrain();
}
state.daemon = { ...daemon };
watchUpgrade();
emit({ kind: 'daemon', envId: forEnv });
},

applyEvent(seq: number, ev: DaemonEvent, forEnv: string): void {
if (forEnv !== envId) return;
if (cursor !== null && seq <= cursor) return;
Expand Down
5 changes: 3 additions & 2 deletions src/renderer/instance-sync.ts
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
/**
* Keeps the instance store in step with the bridge: the instance list,
* the pushed instance events, and the daemon events and snapshots of the
* environment on screen.
* the pushed instance events, and the daemon events, welcomes, and
* snapshots of the environment on screen.
*
* Opening an environment asks main to attach and returns at once; the
* snapshot is read as soon as the attach is up (right away when main is
Expand Down Expand Up @@ -157,6 +157,7 @@ export function initInstanceSync(ctx: InstanceSyncContext) {
bridge.onDaemonEvent((payload) => {
if (payload.envId !== store.envId()) return;
if ('snapshot' in payload) store.applySnapshot(payload.snapshot, payload.envId);
else if ('welcome' in payload) store.applyWelcome(payload.welcome.daemon, payload.welcome.head, payload.envId);
else store.applyEvent(payload.seq, payload.ev, payload.envId);
notify();
});
Expand Down
24 changes: 19 additions & 5 deletions src/renderer/topbar.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,13 @@
* - Chips: "Update available" (the apply dialog, with the changes grouped
* by how they apply), a GitHub warning when the environment's GitHub
* access is not ok, a reconnecting spinner, "Daemon update" when this
* Puck carries a newer daemon, and "Daemon updating" while that runs.
* Puck carries a newer daemon, "Daemon updating" while that runs, and
* "Daemon update failed" when it does not finish.
* - The banner over the content says when the environment cannot be reached
* ("Can't reach build-box. Work continues there; Puck keeps trying."),
* is incompatible, or is not connected, with Reconnect where it helps.
* is incompatible, is not connected, or a daemon update failed. Reconnect
* shows when the attach view offers it; Retry update shows while the
* environment is attached.
*
* Context in, controller out; no DOM lookups.
*/
Expand Down Expand Up @@ -292,7 +295,8 @@ export function initTopbar(ctx: TopbarContext) {
const text = live.github.state === 'expiring' ? 'GitHub access expiring' : live.github.state === 'revoked' ? 'GitHub access revoked' : 'No GitHub access';
els.chips.appendChild(chip('github bad', text, 'The runner keeps GitHub tokens coming from the Puck server; check that the runner is online and Puck is installed on the repositories.'));
}
if (live?.upgrading) els.chips.appendChild(chip('upgrading', 'Daemon updating', live.upgrading === 'drain' ? 'Updating once running turns finish' : 'Updating now'));
if (live?.upgradeError) els.chips.appendChild(chip('daemon bad', 'Daemon update failed', live.upgradeError));
else if (live?.upgrading) els.chips.appendChild(chip('upgrading', 'Daemon updating', live.upgrading === 'drain' ? 'Updating once running turns finish' : 'Updating now'));
else if (info.daemonUpdate && info.attach === 'attached') {
const b = button('tb-chip daemon', 'Daemon update');
b.title = 'This Puck carries a newer environment daemon';
Expand All @@ -303,17 +307,27 @@ export function initTopbar(ctx: TopbarContext) {

function renderBanner(): void {
const view = ctx.attach();
const upgradeError = store.state()?.upgradeError;
els.banner.textContent = '';
const show =
!!upgradeError ||
view.phase === 'unreachable' ||
view.phase === 'incompatible' ||
view.phase === 'detached' ||
view.phase === 'lost' ||
view.phase === 'snapshot-failed';
els.banner.classList.toggle('hidden', !show);
if (!show) return;
els.banner.dataset.phase = view.phase;
els.banner.appendChild(el('span', 'tb-banner-text', view.text));
els.banner.dataset.phase = upgradeError ? 'daemon-update-failed' : view.phase;
els.banner.appendChild(el('span', 'tb-banner-text', upgradeError || view.text));
if (upgradeError && current()?.attach === 'attached') {
const retry = button('btn-ghost', 'Retry update');
retry.addEventListener('click', () => {
const id = store.envId();
if (id) showDaemonDialog(id);
});
els.banner.appendChild(retry);
}
if (view.retry) {
const again = button('btn-ghost', 'Reconnect');
again.addEventListener('click', () => ctx.reconnect());
Expand Down
Loading
Loading