ship-budget resume opens a new meter epoch for a successor thread (GST-1655) (#4088)

* ship-budget resume opens a new meter epoch for a successor thread's counter (GST-1655)

Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com>

* test: ship-budget successor cases exercise lib/ship-budget directly (test-of-test ratchet, GST-1655)

Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com>

* fix: pre-landing review fixes

Review-Finding: p1/red-team#1 reader=red-team
Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com>

* fix: refuse a successor epoch while a handoff child is open (pre-landing review)

Review-Finding: p1/outside-voice#1 reader=outside-voice
Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com>

* fix: successor refusal points at reconcile --return; a reconcile without it writes the child off (GST-1655 fix-check#1)

Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com>

* fix: refuse a successor epoch while a dispatch reservation still has readers on the previous counter (pre-landing review)

Review-Finding: p1/outside-voice#1 reader=outside-voice
Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com>

* docs: changelog fragment names the successor refusals (GST-1655)

Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com>

---------

Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com>

lib/result-codes-budget.ts

@@ -27,7 +27,7 @@ export const BUDGET_RESULT_CODES = {
   BUDGET_READING_REJECTED: {
     anchor: 'budget-reading-rejected',
     summary: 'the reading does not fit the run\'s meter contract: it is lower than the last reading of the same epoch, it was passed to a meter that reads its own source, or --prior-reading is above --reading',
-    fix: 'pass the counter the run started with (the BUDGET line names meter, unit and epoch); a different meter starts a new epoch through gstack-ship-budget resume --meter <m>',
+    fix: 'pass the counter the run started with (the BUDGET line names meter, unit and epoch); a successor thread\'s counter starts a new epoch through gstack-ship-budget resume --meter-source <its jam id> --reading <n> --credits <c>, a different meter through resume --meter <m>',
   },
   BUDGET_METER_UNAVAILABLE: {
     anchor: 'budget-meter-unavailable',
@@ -37,7 +37,7 @@ export const BUDGET_RESULT_CODES = {
   BUDGET_LEDGER_INVALID: {
     anchor: 'budget-ledger-invalid',
     summary: 'the budget ledger or the record\'s gstack-budget block is missing, unparseable, or from another schema version',
-    fix: 'restore it from the Drive checkpoint (gstack-ship-budget resume --run <id>) or rebuild it from the record with resume --record <path> --rebuild-from-record --reading <n>; open reservations are then charged as unknown',
+    fix: 'restore it from the Drive checkpoint (gstack-ship-budget resume --run <id>) or rebuild it from the record with resume --record <path> --rebuild-from-record --reading <n> (plus --credits <c> when the run metered credits); open reservations are then charged as unknown',
   },
   BUDGET_CHECK_SKIPPED: {
     anchor: 'budget-check-skipped',

lib/ship-budget-cli.ts

@@ -238,7 +238,7 @@ function pickMeter(o: Opts): MeterKind | undefined {
 }
 
 export function cmdStart(argv: string[], ctx: Ctx): Outcome {
-  const o = parseArgs(argv, ['token-cap', 'record', 'meter', 'reading', 'credits', 'prior-reading', 'session', 'nonce', 'reserve', 'delegate', 'calibration', 'ledger', 'call-floor', 'mode', 'meter-scope', 'reading-from'], ['subagent']);
+  const o = parseArgs(argv, ['token-cap', 'record', 'meter', 'reading', 'credits', 'prior-reading', 'session', 'nonce', 'reserve', 'delegate', 'calibration', 'ledger', 'call-floor', 'mode', 'meter-scope', 'reading-from', 'meter-source'], ['subagent']);
   const mode = str(o, 'mode') ?? 'admit';
   if (mode !== 'admit' && mode !== 'observe') throw usageError(`--mode '${mode}' is not admit or observe`);
   const scope = str(o, 'meter-scope');
@@ -277,7 +277,7 @@ export function cmdStart(argv: string[], ctx: Ctx): Outcome {
     schema_version: 1, kind: 'gstack-ship-budget', run, dir, file: str(o, 'ledger') ?? path.join(dir, 'budget.json'), record: path.resolve(record),
     created_at: new Date().toISOString(), cap, tail: tokensOpt(o, 'reserve') ?? DEFAULT_TAIL, ...(tokensOpt(o, 'reserve') !== undefined ? { tail_explicit: true } : {}),
     stop_overhead: unitConstants(cal, contract.unit).stop_overhead,
-    delegate, role: 'parent', meter: { contract, epoch: 1, baseline: 0, last: null, ...(sessions ? { sessions, config_dir: claudeConfigDir(ctx.env) } : {}) },
+    delegate, role: 'parent', meter: { contract, epoch: 1, baseline: 0, last: null, ...(sessions ? { sessions, config_dir: claudeConfigDir(ctx.env) } : {}), ...(str(o, 'meter-source') ? { source: str(o, 'meter-source') } : {}) },
     epochs: [], carried_used: 0, estimate_used: 0, unknown: 0, next_id: 1, reservations: [], handoffs: [], cap_raises: [],
     marker: paths.marker, review_log: paths.reviewLog, ...(str(o, 'calibration') ? { calibration: path.resolve(str(o, 'calibration')!) } : {}),
     ...(callFloor !== undefined ? { call_floor: callFloor } : {}), ...(mode === 'observe' ? { mode } : {}),
@@ -435,7 +435,8 @@ export function cmdCalls(argv: string[], ctx: Ctx): Outcome {
 
 /** resume: re-attach after a stop or a machine change; rebuild or restore a lost ledger; journal a cap raise. */
 export function cmdResume(argv: string[], ctx: Ctx): Outcome {
-  const o = parseArgs(argv, ['record', 'run', 'reading', 'credits', 'prior-reading', 'meter', 'session', 'nonce', 'token-cap', 'approved-by', 'words'], ['rebuild-from-record']);
+  const o = parseArgs(argv, ['record', 'run', 'reading', 'credits', 'prior-reading', 'meter', 'session', 'nonce', 'token-cap', 'approved-by', 'words', 'meter-source'], ['rebuild-from-record']);
+  const source = str(o, 'meter-source');
   const cap = str(o, 'token-cap');
   if (cap !== undefined && (!str(o, 'approved-by') || !str(o, 'words'))) throw usageError('resume --token-cap needs --approved-by <who> --words <the owner\'s words>; raising the cap is the owner\'s call');
   const lines: string[] = [];
@@ -453,14 +454,24 @@ export function cmdResume(argv: string[], ctx: Ctx): Outcome {
   }
   const out = mutateLedger(file, ctx.env, l => {
     const meter = pickMeter(o) ?? l.meter.contract.meter;
-    if (meter !== l.meter.contract.meter) {
+    const successor = source !== undefined && source !== l.meter.source;
+    if (successor && meter === 'capy' && str(o, 'reading') === undefined) throw budgetError('BUDGET_READING_MISSING', `resume --meter-source ${source}: a successor's counter needs --reading <its cost.tokens>, the new epoch's baseline`);
+    if (successor && l.credits && str(o, 'credits') === undefined) throw budgetError('BUDGET_READING_MISSING', `resume --meter-source ${source}: run ${l.run} meters credits; pass --credits <its cost.credits>, the new epoch's baseline`);
+    const open = successor ? l.handoffs.filter(h => !h.reconciled_at) : [];
+    if (open.length) throw budgetError('BUDGET_READING_REJECTED', `resume --meter-source ${source}: handoff ${open.map(h => `${h.sha} (steps ${h.from}-${h.to})`).join(', ')} still runs on ${l.meter.source ?? 'the previous'} counter, which the new epoch cannot see; wait for its RETURN, reconcile it with a reading of that counter (${open.map(h => `gstack-ship-budget reconcile --run ${l.run} --return ${h.return_file} --reading <its cost.tokens>`).join('; ')}), then resume. A reconcile without --return writes the child off as interrupted, and its later spend is never charged`);
+    const inFlight = successor ? l.reservations.filter(r => r.kind === 'dispatch' && (r.state === 'open' || r.state === 'consumed')) : [];
+    if (inFlight.length) throw budgetError('BUDGET_READING_REJECTED', `resume --meter-source ${source}: dispatch reservation ${inFlight.map(r => `${r.id} (step ${r.step}, ${r.state})`).join(', ')} still has readers on ${l.meter.source ?? 'the previous'} counter, which the new epoch cannot see; wait for them to settle, settle each with a reading of that counter (${inFlight.map(r => `gstack-ship-budget settle --run ${l.run} --reservation ${r.id} --dispatched-at <its reading when the readers started> --reading <its cost.tokens>`).join('; ')}), then resume`);
+    if (meter !== l.meter.contract.meter || successor) {
+      const previous = l.meter.source ?? 'unrecorded';
       const sessions = meter === 'claude' ? bindClaude(o, ctx.env) : undefined;
-      newEpoch(l, METER_CONTRACTS[meter], meter === 'capy' ? tokensOpt(o, 'reading') ?? 0 : 0);
+      newEpoch(l, METER_CONTRACTS[meter], meter === 'capy' ? tokensOpt(o, 'reading') ?? 0 : 0,
+        successor ? { source, ...(str(o, 'credits') === undefined ? {} : { credits: parseCredits(str(o, 'credits'), '--credits') }) } : undefined);
       if (sessions) { l.meter.sessions = sessions; l.meter.config_dir = claudeConfigDir(ctx.env); try { fs.unlinkSync(path.join(l.dir, 'claude-meter.json')); } catch {} }
       const call = readMeter(l, { ...o, ...(meter === 'capy' ? {} : { reading: undefined as any }) }, false);
       if (meter === 'claude') { l.meter.baseline = l.meter.last ?? 0; }
       void call;
-      lines.push(`EPOCH: ${l.meter.epoch} meter=${meter} unit=${l.meter.contract.unit} carried_used=${l.carried_used}`);
+      lines.push(`EPOCH: ${l.meter.epoch} meter=${meter} unit=${l.meter.contract.unit} carried_used=${l.carried_used}`
+        + (successor ? `${l.credits?.carried === undefined ? '' : ` carried_credits=${l.credits.carried}`} source=${source} previous_source=${previous}` : ''));
     } else readMeter(l, o, false);
     if (cap !== undefined) {
       const to = cap === 'none' ? null : parseTokens(cap, '--token-cap');
@@ -482,12 +493,15 @@ function rebuildLedger(file: string, record: string | undefined, ctx: Ctx, o: Op
   const meter = b.meter as MeterKind;
   const reading = tokensOpt(o, 'reading');
   if (meter === 'capy' && reading === undefined) throw budgetError('BUDGET_READING_MISSING', 'resume --rebuild-from-record on meter=capy');
+  const credits = str(o, 'credits') === undefined ? undefined : parseCredits(str(o, 'credits'), '--credits');
+  if (b.credits !== undefined && credits === undefined) throw budgetError('BUDGET_READING_MISSING', `resume --rebuild-from-record: run ${b.run} metered credits (credits=${b.credits}); pass --credits <cost.credits>, the new epoch's baseline`);
   const num = (k: string) => Number(b[k] ?? 0) || 0;
   const l: Ledger = {
     schema_version: 1, kind: 'gstack-ship-budget', run: b.run, dir: path.dirname(file), file, record: path.resolve(record!),
     created_at: new Date().toISOString(), cap: b.cap === 'none' ? null : num('cap'), tail: num('tail') || DEFAULT_TAIL,
     stop_overhead: unitConstants(loadCalibration(), METER_CONTRACTS[meter].unit).stop_overhead, delegate: 'auto', role: b.role === 'child' ? 'child' : 'parent',
-    meter: { contract: METER_CONTRACTS[meter], epoch: num('epoch') + 1, baseline: meter === 'capy' ? reading! : 0, last: meter === 'capy' ? reading! : null },
+    meter: { contract: METER_CONTRACTS[meter], epoch: num('epoch') + 1, baseline: meter === 'capy' ? reading! : 0, last: meter === 'capy' ? reading! : null, ...(str(o, 'meter-source') ? { source: str(o, 'meter-source') } : {}) },
+    ...(credits === undefined ? {} : { credits: { baseline: credits, last: credits, ...(b.credits === undefined ? {} : { carried: num('credits') }) } }),
     epochs: [], carried_used: num('used'), estimate_used: 0, unknown: num('unknown') + num('reserved'), next_id: 1, reservations: [], handoffs: [], cap_raises: [],
     ...(() => { try { return { marker: reviewPaths(ctx.cwd, ctx.env).marker, review_log: reviewPaths(ctx.cwd, ctx.env).reviewLog }; } catch { return {}; } })(),
     ...(b.mode === 'observe' ? { mode: 'observe' as const } : {}), ...(b.meter_scope === 'tree' ? { meter_scope: 'tree' as const } : {}),

lib/ship-budget.ts

@@ -11,7 +11,8 @@
  * lib/spend-ledger.ts's withFileLock, so lib/autoplan-budget.ts's USD ledger is
  * untouched. Meters are versioned contracts (unit, host field, scope,
  * descendants, epoch); a reading lower than the last one of its epoch is
- * rejected and a meter change opens a new epoch that keeps prior usage. Under a
+ * rejected; a meter change, or a successor thread's counter (resume --meter-source,
+ * GST-1655), opens a new epoch that keeps prior usage. Under a
  * cap, start writes a per-branch marker next to the review log so
  * gstack-review-log --start refuses a dispatch without an unspent reservation;
  * without a cap no marker exists and review-log behaves as before. Commands:
@@ -81,8 +82,9 @@ export interface Ledger {
   schema_version: 1; kind: 'gstack-ship-budget'; run: string; dir: string; file: string; record?: string; created_at: string;
   cap: number | null; tail: number; stop_overhead: number; delegate: 'auto' | 'never'; role: 'parent' | 'child';
   parent?: { run: string; ledger: string; handoff_sha: string; from: string; to: string; return_file: string };
-  meter: { contract: MeterContract; epoch: number; baseline: number; last: number | null; last_at?: string; sessions?: string[]; config_dir?: string };
-  epochs: Array<{ epoch: number; contract: MeterContract; baseline: number; last: number | null; used: number }>;
+  /** source: the thread whose counter the readings come from (--meter-source); a different one on resume is a successor. */
+  meter: { contract: MeterContract; epoch: number; baseline: number; last: number | null; last_at?: string; sessions?: string[]; config_dir?: string; source?: string };
+  epochs: Array<{ epoch: number; contract: MeterContract; baseline: number; last: number | null; used: number; source?: string; credits?: number }>;
   carried_used: number; estimate_used: number; unknown: number; next_id: number;
   reservations: Reservation[]; handoffs: Handoff[];
   cap_raises: Array<{ from: number | null; to: number | null; by: string; words: string; at: string }>;
@@ -94,8 +96,8 @@ export interface Ledger {
   parent_metered?: boolean;
   /** observe (ship/DRIVER.md): readings are recorded by note and the cap is one hard stop; no reservation, admission or delegation. */
   mode?: 'admit' | 'observe';
-  /** Capy cost.credits readings beside the token meter: baseline is the first reading, last the newest. */
-  credits?: { baseline: number; last: number };
+  /** Capy cost.credits readings beside the token meter: baseline is the first reading of its epoch, last the newest, carried what earlier sources spent. */
+  credits?: { baseline: number; last: number; carried?: number };
   notes?: Note[];
   /** tree: the reading is the root thread's counter (a subagent cannot read its own jam id), so every cost line carries TREE_METER. */
   meter_scope?: 'tree';
@@ -415,10 +417,10 @@ export function parseCredits(raw: string | undefined, flag: string): number {
 export function applyCredits(l: Ledger, value: number | undefined): void {
   if (value === undefined) return;
   if (l.credits && value < l.credits.last) throw budgetError('BUDGET_READING_REJECTED', `--credits ${value} is below the last credits reading ${l.credits.last}`);
-  l.credits = { baseline: l.credits?.baseline ?? value, last: value };
+  l.credits = { ...l.credits, baseline: l.credits?.baseline ?? value, last: value };
 }
 
-export const creditsUsed = (l: Ledger): number | null => (l.credits ? Math.round((l.credits.last - l.credits.baseline) * 100) / 100 : null);
+export const creditsUsed = (l: Ledger): number | null => (l.credits ? Math.round(((l.credits.carried ?? 0) + l.credits.last - l.credits.baseline) * 100) / 100 : null);
 
 export function budgetLine(l: Ledger, f: { call: number | null; next: string; projected: number; verdict: string }): string {
   const t = tally(l);
@@ -469,13 +471,21 @@ export function applyReading(l: Ledger, value: number | null, source: 'flag' | '
   l.meter.last_at = new Date().toISOString();
 }
 
-/** A meter change: close the epoch (its usage carries) and start the next at the new baseline. */
-export function newEpoch(l: Ledger, contract: MeterContract, baseline: number): void {
+/**
+ * A meter change: close the epoch (its usage carries) and start the next at the new baseline.
+ * A successor (another thread's counter) also closes the credits epoch: what they used carries
+ * and its credits reading is the new baseline.
+ */
+export function newEpoch(l: Ledger, contract: MeterContract, baseline: number, successor?: { source: string; credits?: number }): void {
   const metered = l.meter.contract.meter === 'estimate' ? l.estimate_used : Math.max(0, (l.meter.last ?? l.meter.baseline) - l.meter.baseline);
-  l.epochs.push({ epoch: l.meter.epoch, contract: l.meter.contract, baseline: l.meter.baseline, last: l.meter.last, used: metered });
+  const credits = successor ? creditsUsed(l) : null;
+  l.epochs.push({ epoch: l.meter.epoch, contract: l.meter.contract, baseline: l.meter.baseline, last: l.meter.last, used: metered,
+    ...(l.meter.source ? { source: l.meter.source } : {}), ...(credits === null ? {} : { credits }) });
   l.carried_used += metered;
   l.estimate_used = 0;
-  l.meter = { contract, epoch: l.meter.epoch + 1, baseline, last: baseline };
+  const source = successor?.source ?? l.meter.source;
+  l.meter = { contract, epoch: l.meter.epoch + 1, baseline, last: baseline, ...(source ? { source } : {}) };
+  if (credits !== null && successor?.credits !== undefined) l.credits = { baseline: successor.credits, last: successor.credits, carried: credits };
 }
 
 const BLOCK = /```gstack-budget\n[\s\S]*?\n```\n?/;