Work through the remaining correctness findings from the static-analysis sweep

apps/alerting/src/worker.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "@effect/vitest"
-import { Cause, Effect, Exit, Layer } from "effect"
+import { Cause, Effect, Exit, Fiber, Latch, Layer } from "effect"
 import { env as workerEnv } from "../test/stubs/cloudflare-workers"
 import { buildLayer, catchTickFailure, selectScheduledProgram, type ScheduledTickPrograms } from "./worker"
 
@@ -39,6 +39,37 @@ describe("alerting Effect root", () => {
 		})
 	}
 
+	it.effect("never starts fixVerification before the error tick completes", () => {
+		// The adversarial schedule for concurrency 2: alert and escalation are
+		// instant, so both finish while error is still running. Listing the four
+		// ticks side by side would hand the freed slot to fixVerification here —
+		// only the explicit error→fixVerification chain keeps the ordering.
+		const calls: Array<string> = []
+		const record = (name: string) => Effect.sync(() => calls.push(name)).pipe(Effect.asVoid)
+		return Effect.gen(function* () {
+			const errorGate = yield* Latch.make(false)
+			const ticks = {
+				alert: record("alert"),
+				anomaly: record("anomaly"),
+				cloudflareAnalytics: record("cloudflareAnalytics"),
+				digest: record("digest"),
+				error: errorGate.await.pipe(Effect.andThen(record("error"))),
+				escalation: record("escalation"),
+				fixVerification: record("fixVerification"),
+				planetScale: record("planetScale"),
+				serviceMapRollup: record("serviceMapRollup"),
+			} satisfies ScheduledTickPrograms
+			const fiber = yield* Effect.forkChild(selectScheduledProgram("* * * * *", ticks))
+			// Let alert and escalation run to completion while error is held open.
+			yield* Effect.yieldNow
+			yield* Effect.yieldNow
+			expect(calls).not.toContain("fixVerification")
+			yield* errorGate.open
+			yield* Fiber.join(fiber)
+			expect(calls.indexOf("error")).toBeLessThan(calls.indexOf("fixVerification"))
+		})
+	})
+
 	it.effect("fails closed for an unknown cron", () => {
 		const calls: Array<string> = []
 		const tick = (name: string) => Effect.sync(() => calls.push(name)).pipe(Effect.asVoid)

apps/alerting/src/worker.ts

@@ -461,10 +461,12 @@ export const selectScheduledProgram = <R>(
 		Match.when("*/15 * * * *", () => ticks.digest),
 		Match.when("0 * * * *", () => ticks.serviceMapRollup),
 		Match.when("* * * * *", () =>
-			// `fixVerification` runs after `error` in the same group so a window that
-			// this minute's error tick just refuted is already settled when the
-			// verification tick looks at it — one fewer wasted agent start.
-			Effect.all([ticks.alert, ticks.error, ticks.escalation, ticks.fixVerification], {
+			// `fixVerification` is chained onto `error` rather than listed beside it:
+			// a window this minute's error tick just refuted must already be settled
+			// when the verification tick looks at it, and array order under bounded
+			// concurrency does not promise that — alert and escalation finishing
+			// first would have started fixVerification while error still ran.
+			Effect.all([ticks.alert, Effect.andThen(ticks.error, ticks.fixVerification), ticks.escalation], {
 				concurrency: 2,
 				discard: true,
 			}),

apps/api/src/chat/loop/turn.test.ts

@@ -1113,3 +1113,57 @@ describe("runChatTurn closing submit", () => {
 		}),
 	)
 })
+
+describe("runChatTurn duplicate completion calls", () => {
+	it.live("dispatches only the first of duplicate completion calls in an ordinary step", () =>
+		Effect.gen(function* () {
+			// The completion is an exactly-once output channel: a response that
+			// carries it twice must not execute two competing submits.
+			const record: Array<unknown> = []
+			const result = yield* collect(
+				[
+					[
+						toolCall("c1", SUBMIT, { claim: "first" }),
+						toolCall("c2", SUBMIT, { claim: "second" }),
+						finish("tool-calls"),
+					],
+				],
+				{ completion: submitCompletion(record) },
+			)
+
+			assert.deepEqual(record, [{ claim: "first" }])
+			const submitEvents = result.events.filter(
+				(event) => event.type === "tool-call" && event.name === SUBMIT,
+			)
+			assert.lengthOf(submitEvents, 1)
+			const end = terminal(result.events)[0]
+			assert.equal(end?.type === "turn-end" ? end.reason : undefined, "stop")
+		}),
+	)
+
+	it.live("dispatches only the first duplicate on the forced closing step", () =>
+		Effect.gen(function* () {
+			const record: Array<unknown> = []
+			// Prose first step forces the submit closing step; the model then
+			// ignores "exactly one call" and emits the completion twice.
+			const result = yield* collect(
+				[
+					[textDelta("thinking..."), finish()],
+					[
+						toolCall("c1", SUBMIT, { claim: "first" }),
+						toolCall("c2", SUBMIT, { claim: "second" }),
+						finish("tool-calls"),
+					],
+				],
+				{ completion: submitCompletion(record) },
+			)
+
+			assert.deepEqual(record, [{ claim: "first" }])
+			const submitEvents = result.events.filter(
+				(event) => event.type === "tool-call" && event.name === SUBMIT,
+			)
+			assert.lengthOf(submitEvents, 1)
+			assert.lengthOf(terminal(result.events), 1)
+		}),
+	)
+})

apps/api/src/chat/loop/turn.ts

@@ -477,9 +477,20 @@ const runStep = (
 					})
 				}
 
+				// The completion is an exactly-once output channel: a response carrying
+				// duplicate completion calls must not run it twice (competing submits
+				// would leave the report reflecting one payload and its side effects
+				// another). Keep the first and drop the rest before any dispatch.
+				let completionSeen = false
 				const calls = response.events
 					.filter(LLMEvent.is.toolCall)
 					.filter((call) => !call.providerExecuted)
+					.filter((call) => {
+						if (call.name !== input.completion?.name) return true
+						if (completionSeen) return false
+						completionSeen = true
+						return true
+					})
 				const finishReason = response.finishReason?.normalized
 				const providerFailure = response.events.find(LLMEvent.is.providerError)
 

apps/api/src/platform/Apns.test.ts

@@ -0,0 +1,45 @@
+import { assert, describe, it } from "@effect/vitest"
+import { Effect } from "effect"
+import { makeSingleFlightTokenCache } from "./Apns"
+
+describe("makeSingleFlightTokenCache", () => {
+	it.live("many concurrent reads on a cold cache mint exactly once", () =>
+		Effect.gen(function* () {
+			let mints = 0
+			// The mint suspends (as the real ECDSA sign does), so without
+			// single-flight every concurrent fiber would observe the empty cache
+			// and mint its own token.
+			const mint = Effect.gen(function* () {
+				mints += 1
+				yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 5)))
+				return `token-${mints}`
+			})
+			const currentToken = yield* makeSingleFlightTokenCache(60_000, mint)
+			const tokens = yield* Effect.all(
+				Array.from({ length: 8 }, () => currentToken),
+				{ concurrency: "unbounded" },
+			)
+			assert.strictEqual(mints, 1)
+			assert.deepStrictEqual(
+				tokens,
+				Array.from({ length: 8 }, () => "token-1"),
+			)
+		}),
+	)
+
+	it.live("mints again only after the TTL expires", () =>
+		Effect.gen(function* () {
+			let mints = 0
+			const mint = Effect.sync(() => {
+				mints += 1
+				return `token-${mints}`
+			})
+			const currentToken = yield* makeSingleFlightTokenCache(10, mint)
+			assert.strictEqual(yield* currentToken, "token-1")
+			assert.strictEqual(yield* currentToken, "token-1")
+			yield* Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 15)))
+			assert.strictEqual(yield* currentToken, "token-2")
+			assert.strictEqual(mints, 2)
+		}),
+	)
+})

apps/api/src/platform/Apns.ts

@@ -1,4 +1,4 @@
-import { Clock, Context, Effect, Layer, Option, Redacted, Ref, Schema } from "effect"
+import { Clock, Context, Effect, Layer, Option, Redacted, Ref, Schema, Semaphore } from "effect"
 import { FetchHttpClient, HttpClient, HttpClientRequest } from "effect/unstable/http"
 import type { MobilePushEnvironment } from "@maple/domain/http"
 import { Env, type EnvConfig } from "./Env"
@@ -198,6 +198,48 @@ const resolveConfig = (env: EnvConfig): ApnsConfig | null => {
 
 const decodeReason = Schema.decodeUnknownOption(Schema.Struct({ reason: Schema.String }))
 
+/**
+ * TTL cache around `mint` with single-flight refresh. MobilePushService fans
+ * out up to eight concurrent sends, so a cold or expired cache would otherwise
+ * mint one JWT per fiber — and Apple throttles clients that update provider
+ * tokens too often (`TooManyProviderTokenUpdates`). One permit serializes the
+ * stale path; waiters re-check the cache and reuse the winner's token.
+ * Exported for the concurrency test only.
+ */
+export const makeSingleFlightTokenCache = <E, R>(
+	ttlMs: number,
+	mint: Effect.Effect<string, E, R>,
+): Effect.Effect<Effect.Effect<string, E, R>> =>
+	Ref.make(Option.none<{ readonly value: string; readonly mintedAtMs: number }>()).pipe(
+		Effect.map((cached) => {
+			const lock = Semaphore.makeUnsafe(1)
+			// A cached token counts only while younger than the TTL; expiry is
+			// absence, not a null to compare against.
+			const freshValue = (nowMs: number) =>
+				Ref.get(cached).pipe(
+					Effect.map(Option.filter((entry) => nowMs - entry.mintedAtMs < ttlMs)),
+					Effect.map(Option.map((entry) => entry.value)),
+				)
+			return Effect.gen(function* () {
+				const nowMs = yield* Clock.currentTimeMillis
+				const entry = yield* freshValue(nowMs)
+				if (Option.isSome(entry)) return entry.value
+				return yield* lock.withPermits(1)(
+					Effect.gen(function* () {
+						// Double-checked: a fiber that waited here usually finds the
+						// winner's fresh token and must not mint another.
+						const innerNowMs = yield* Clock.currentTimeMillis
+						const latest = yield* freshValue(innerNowMs)
+						if (Option.isSome(latest)) return latest.value
+						const value = yield* mint
+						yield* Ref.set(cached, Option.some({ value, mintedAtMs: innerNowMs }))
+						return value
+					}),
+				)
+			})
+		}),
+	)
+
 export class ApnsClient extends Context.Service<ApnsClient, ApnsClientApi>()(
 	"@maple/api/platform/ApnsClient",
 	{
@@ -237,11 +279,6 @@ export class ApnsClient extends Context.Service<ApnsClient, ApnsClientApi>()(
 				}),
 			)
 
-			const cachedToken = yield* Ref.make<{
-				readonly value: string
-				readonly mintedAtMs: number
-			} | null>(null)
-
 			const mintToken = Effect.fn("ApnsClient.mintToken")(function* () {
 				const nowMs = yield* Clock.currentTimeMillis
 				const header = base64UrlString(JSON.stringify({ alg: "ES256", kid: config.keyId }))
@@ -260,17 +297,10 @@ export class ApnsClient extends Context.Service<ApnsClient, ApnsClientApi>()(
 						),
 					catch: (cause) => new ApnsError({ message: "APNs JWT signing failed", cause }),
 				})
-				const value = `${signingInput}.${base64UrlBytes(signature)}`
-				yield* Ref.set(cachedToken, { value, mintedAtMs: nowMs })
-				return value
+				return `${signingInput}.${base64UrlBytes(signature)}`
 			})
 
-			const currentToken = Effect.gen(function* () {
-				const nowMs = yield* Clock.currentTimeMillis
-				const cached = yield* Ref.get(cachedToken)
-				if (cached !== null && nowMs - cached.mintedAtMs < TOKEN_TTL_MS) return cached.value
-				return yield* mintToken()
-			})
+			const currentToken = yield* makeSingleFlightTokenCache(TOKEN_TTL_MS, mintToken())
 
 			const send = Effect.fn("ApnsClient.send")(function* (push: ApnsPush) {
 				yield* Effect.annotateCurrentSpan({