chat

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136// Stock release decisions over public snapshots. Delivery integrity stays in the
// host; this versioned, opaque state can be replaced by another scheduler.
import * as z from "zod";
import type { InputSnapshot, PendingInput, UsageRecovery } from "convergence";

export const recoveryWait = z.object({
  runId: z.string(),
  stoppedAt: z.string(),
  identity: z.string().optional(),
  scope: z.string().optional(),
  resetAt: z.string().optional(),
  reason: z.string(),
  status: z.enum(["armed", "paused", "attempting", "failed"]),
  attemptKey: z.string().optional(),
  inputId: z.string().optional(),
  generated: z.boolean().optional(),
  checkedKey: z.string().optional(),
});
export const policySchema = z.object({
  version: z.literal(1),
  generation: z.number(),
  handledRun: z.string().optional(),
  releasedRun: z.string().optional(),
  pause: z.string().optional(),
  wait: recoveryWait.optional(),
});
export type PolicyState = z.infer<typeof policySchema>;
export type RecoveryWait = z.infer<typeof recoveryWait>;
export const preferencesSchema = z.object({
  autoSendAfterLimit: z.record(z.string(), z.boolean()).default({}),
  generateContinue: z.boolean().default(false),
});
export type Preferences = z.infer<typeof preferencesSchema>;
export const defaults: Preferences = { autoSendAfterLimit: {}, generateContinue: false };
export const autoSend = (preferences: Preferences, agentId: string) => preferences.autoSendAfterLimit[agentId] !== false;
export const editable = (input: PendingInput) => input.state === "held" || input.state === "rejected";
export const uncertain = (snapshot: InputSnapshot) =>
  snapshot.inputs.some((input) => ["dispatching", "delivery_unknown", "native_admitted"].includes(input.state));
export const pending = (snapshot: InputSnapshot) => snapshot.inputs.filter((input) => input.state !== "observed_consumed");
export const firstQueued = (snapshot: InputSnapshot) =>
  snapshot.inputs.find((input) => input.intent === "queue" && input.state === "held" && !input.pauseReason);

export function policyOf(snapshot: InputSnapshot): PolicyState {
  const result = policySchema.safeParse(snapshot.policyState);
  if (result.success && result.data.generation === snapshot.generation) return result.data;
  // A changed attachment/generation is an invalidation seam, not a handoff
  // implementation. Neither a previous success nor a source timer can release it.
  return {
    version: 1,
    generation: snapshot.generation,
    handledRun: snapshot.lastRun?.id,
    releasedRun: snapshot.lastRun?.id,
    ...(snapshot.policyState != null ? { pause: "Attachment changed; send explicitly to resume" } : {}),
  };
}

export function settlePolicy(snapshot: InputSnapshot, state: PolicyState): PolicyState {
  const run = snapshot.lastRun;
  if (!run || run.generation !== snapshot.generation || run.id === state.handledRun) return state;
  const next = { ...state, handledRun: run.id };
  // Simulated cancellation is owned by the host reservation, never a FIFO release.
  if (run.steering) return next;
  if (run.outcome.status === "completed") return { ...next, wait: undefined, releasedRun: firstQueued(snapshot) ? state.releasedRun : run.id };
  if (run.outcome.status === "failed" && run.usageLimit && !state.pause) {
    const evidence = run.usageLimit;
    return {
      ...next,
      wait: {
        runId: run.id,
        stoppedAt: run.stoppedAt,
        identity: evidence.identity,
        scope: evidence.scope,
        resetAt: validReset(evidence.resetsAt, run.stoppedAt) ? evidence.resetsAt : undefined,
        reason: evidence.reason,
        status: "armed",
      },
    };
  }
  return { ...next, pause: state.pause ?? (run.outcome.status === "cancelled" ? "Stopped by you" : "Run failed"), wait: state.wait ? { ...state.wait, status: "paused" } : undefined };
}

export function fifoCandidate(snapshot: InputSnapshot, state: PolicyState): PendingInput | undefined {
  const run = snapshot.lastRun;
  if (snapshot.running || snapshot.settling || uncertain(snapshot) || state.pause || state.wait || !run || run.steering ||
      run.generation !== snapshot.generation || run.outcome.status !== "completed" || state.releasedRun === run.id) return;
  return firstQueued(snapshot);
}

export function validReset(reset: string | undefined, stoppedAt: string): boolean {
  return !!reset && Number.isFinite(Date.parse(reset)) && Date.parse(reset) > Date.parse(stoppedAt);
}
export function scopeOf(snapshot: InputSnapshot, wait: RecoveryWait): string | undefined {
  // Identity is never fabricated from a percentage or agent name. With no
  // pool reported, conservatively coordinate only that agent/credential pair.
  return wait.identity ? JSON.stringify([snapshot.agentId, wait.identity, wait.scope ?? "credential"]) : undefined;
}
export function waitKey(wait: RecoveryWait): string {
  return JSON.stringify([wait.runId, wait.identity, wait.scope, wait.resetAt]);
}
export function matching(wait: RecoveryWait, evidence: UsageRecovery): boolean {
  return !!wait.identity && evidence.identity === wait.identity && evidence.scope === wait.scope;
}
export function fresh(evidence: UsageRecovery, wait: RecoveryWait, checkedAt: number): boolean {
  const observed = Date.parse(evidence.observedAt);
  return observed >= Date.parse(wait.stoppedAt) && observed >= checkedAt - 5000 && observed <= checkedAt + 5000;
}
export type RecoveryDecision = { kind: "send"; speculative: boolean } | { kind: "hold"; reason: string; resetAt?: string } | { kind: "invalidate"; reason: string };
export function recoveryDecision(wait: RecoveryWait, evidence: UsageRecovery | undefined, now: number, restored: boolean): RecoveryDecision {
  if (!wait.identity) return { kind: "hold", reason: "Account identity unavailable; retry manually" };
  if (evidence?.identity && !matching(wait, evidence)) return { kind: "invalidate", reason: "Account or quota scope changed; retry manually" };
  const usable = evidence && matching(wait, evidence) && fresh(evidence, wait, now) ? evidence : undefined;
  if (restored && !usable) return { kind: "hold", reason: "Account could not be reconciled after restart; retry manually" };
  if (usable?.availability === "blocked") return { kind: "hold", reason: usable.reason, resetAt: validReset(usable.resetsAt, wait.stoppedAt) ? usable.resetsAt : undefined };
  if (wait.attemptKey === waitKey(wait)) return { kind: "hold", reason: "Recovery retry already attempted; retry manually" };
  if (usable?.availability === "allowed") return { kind: "send", speculative: false };
  if (validReset(wait.resetAt, wait.stoppedAt) && Date.parse(wait.resetAt ?? "") <= now) return { kind: "send", speculative: true };
  return { kind: "hold", reason: "Reset time unavailable; retry manually" };
}

export function orderedWaits(snapshots: readonly InputSnapshot[]): InputSnapshot[] {
  return snapshots.filter((snapshot) => policyOf(snapshot).wait?.status === "armed")
    .sort((a, b) => {
      const left = policyOf(a).wait, right = policyOf(b).wait;
      return (left?.stoppedAt ?? "").localeCompare(right?.stoppedAt ?? "") || a.chatId.localeCompare(b.chatId) ||
        (pending(a)[0]?.id ?? "").localeCompare(pending(b)[0]?.id ?? "");
    });
}

export function inputLabel(input: PendingInput): string {
  const prefix = input.origin === "generated_continue" ? "Generated Continue" : input.intent === "steer" ? "Steer" : input.intent === "send" ? "Send" : "Queued";
  const delivery = {
    held: "awaiting delivery", rejected: "not admitted", dispatching: "submitting", delivery_unknown: "delivery unknown — do not replay",
    native_admitted: "admitted — pickup not yet observed", observed_consumed: "pickup observed",
  }[input.state];
  return `${prefix} · ${delivery}${input.pauseReason ? ` · ${input.pauseReason}` : ""}`;
}