Added token clamping, TodoWrite plan events, and ProviderStatusCache service

Adapted from upstream t3code pingdotgg#1943 and pingdotgg#1541. normalizeClaudeTokenUsage now clamps
usedTokens to contextWindow when it exceeds it, and includes totalProcessedTokens in
the snapshot when it differs. Added isTodoTool and extractPlanStepsFromTodoInput to
emit turn.plan.updated events during TodoWrite input streaming and content_block_stop.
Created ProviderStatusCache Effect service (in-memory Ref-backed cache) for future use.

Author: youpele52

apps/server/src/provider/Layers/ClaudeAdapter.stream.handlers.ts

@@ -8,6 +8,8 @@ import {
   asRuntimeItemId,
   classifyToolItemType,
   extractExitPlanModePlan,
+  extractPlanStepsFromTodoInput,
+  isTodoTool,
   nativeProviderRefs,
   streamKindFromDeltaType,
   summarizeToolRequest,
@@ -169,6 +171,27 @@ export const makeMessageHandlers = (deps: MessageHandlerDeps) => {
             payload: message,
           },
         });
+
+        if (isTodoTool(nextTool.toolName) && parsedInput) {
+          const planSteps = extractPlanStepsFromTodoInput(parsedInput);
+          if (planSteps.length > 0 && context.turnState) {
+            const planStamp = yield* makeEventStamp();
+            yield* offerRuntimeEvent({
+              type: "turn.plan.updated",
+              eventId: planStamp.eventId,
+              provider: PROVIDER,
+              createdAt: planStamp.createdAt,
+              threadId: context.session.threadId,
+              turnId: context.turnState.turnId,
+              payload: {
+                plan: planSteps,
+              },
+              providerRefs: nativeProviderRefs(context, {
+                providerItemId: nextTool.itemId,
+              }),
+            });
+          }
+        }
       }
       return;
     }
@@ -256,6 +279,25 @@ export const makeMessageHandlers = (deps: MessageHandlerDeps) => {
       if (!tool) {
         return;
       }
+
+      if (isTodoTool(tool.toolName) && tool.input && typeof tool.input === "object") {
+        const planSteps = extractPlanStepsFromTodoInput(tool.input as Record<string, unknown>);
+        if (planSteps.length > 0 && context.turnState) {
+          const planStamp = yield* makeEventStamp();
+          yield* offerRuntimeEvent({
+            type: "turn.plan.updated",
+            eventId: planStamp.eventId,
+            provider: PROVIDER,
+            createdAt: planStamp.createdAt,
+            threadId: context.session.threadId,
+            turnId: context.turnState.turnId,
+            payload: {
+              plan: planSteps,
+            },
+            providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }),
+          });
+        }
+      }
     }
   });
 

apps/server/src/provider/Layers/ClaudeAdapter.utils.ts

@@ -9,6 +9,8 @@ import {
   type CanonicalRequestType,
   ClaudeCodeEffort,
   RuntimeRequestId,
+  type RuntimePlanStepStatus,
+  RUNTIME_PLAN_STEP_STATUSES,
   type ThreadTokenUsageSnapshot,
   ThreadId,
   type TurnId,
@@ -168,19 +170,28 @@ export function normalizeClaudeTokenUsage(
       ? record.output_tokens
       : 0;
   const derivedUsedTokens = inputTokens + outputTokens;
+  const totalProcessedTokens = directUsedTokens;
   const usedTokens = directUsedTokens ?? (derivedUsedTokens > 0 ? derivedUsedTokens : undefined);
   if (usedTokens === undefined || usedTokens <= 0) {
     return undefined;
   }
 
+  const hasContextWindow =
+    typeof contextWindow === "number" && Number.isFinite(contextWindow) && contextWindow > 0;
+  const clampedUsedTokens =
+    hasContextWindow && usedTokens > contextWindow ? contextWindow : usedTokens;
+  const shouldIncludeTotalProcessed =
+    totalProcessedTokens !== undefined &&
+    totalProcessedTokens > 0 &&
+    totalProcessedTokens !== clampedUsedTokens;
+
   return {
-    usedTokens,
-    lastUsedTokens: usedTokens,
+    usedTokens: clampedUsedTokens,
+    lastUsedTokens: clampedUsedTokens,
     ...(inputTokens > 0 ? { inputTokens } : {}),
     ...(outputTokens > 0 ? { outputTokens } : {}),
-    ...(typeof contextWindow === "number" && Number.isFinite(contextWindow) && contextWindow > 0
-      ? { maxTokens: contextWindow }
-      : {}),
+    ...(hasContextWindow ? { maxTokens: contextWindow } : {}),
+    ...(shouldIncludeTotalProcessed ? { totalProcessedTokens } : {}),
     ...(typeof record.tool_uses === "number" && Number.isFinite(record.tool_uses)
       ? { toolUses: record.tool_uses }
       : {}),
@@ -292,6 +303,25 @@ export function isReadOnlyToolName(toolName: string): boolean {
   );
 }
 
+export function isTodoTool(toolName: string): boolean {
+  return toolName === "TodoWrite";
+}
+
+export function extractPlanStepsFromTodoInput(
+  input: Record<string, unknown>,
+): ReadonlyArray<{ step: string; status: RuntimePlanStepStatus }> {
+  const todos = Array.isArray(input.todos) ? input.todos : [];
+  return todos
+    .filter((t): t is Record<string, unknown> => !!t && typeof t === "object")
+    .map((t) => ({
+      step:
+        typeof t.content === "string" && t.content.trim().length > 0 ? t.content.trim() : "Task",
+      status: (RUNTIME_PLAN_STEP_STATUSES as readonly string[]).includes(t.status as string)
+        ? (t.status as RuntimePlanStepStatus)
+        : ("pending" as RuntimePlanStepStatus),
+    }));
+}
+
 export function classifyRequestType(toolName: string): CanonicalRequestType {
   if (isReadOnlyToolName(toolName)) {
     return "file_read_approval";

apps/server/src/provider/Layers/ProviderStatusCache.ts

@@ -0,0 +1,23 @@
+import type { ThreadId } from "@bigcode/contracts";
+import { Effect, Layer, Ref } from "effect";
+
+import { ProviderStatusCache, type ProviderStatusEntry } from "../Services/ProviderStatusCache.ts";
+
+export const ProviderStatusCacheLive = Layer.effect(
+  ProviderStatusCache,
+  Effect.gen(function* () {
+    const cache = yield* Ref.make(new Map<ThreadId, ProviderStatusEntry>());
+
+    return {
+      get: (threadId: ThreadId) => Ref.get(cache).pipe(Effect.map((m) => m.get(threadId))),
+      set: (threadId: ThreadId, entry: ProviderStatusEntry) =>
+        Ref.update(cache, (m) => new Map(m).set(threadId, entry)),
+      delete: (threadId: ThreadId) =>
+        Ref.update(cache, (m) => {
+          const next = new Map(m);
+          next.delete(threadId);
+          return next;
+        }),
+    };
+  }),
+);

apps/server/src/provider/Services/ProviderStatusCache.ts

@@ -0,0 +1,20 @@
+import type { ProviderKind, ThreadId } from "@bigcode/contracts";
+import { ServiceMap } from "effect";
+import type { Effect } from "effect";
+
+export interface ProviderStatusEntry {
+  readonly providerName: ProviderKind;
+  readonly status: "initializing" | "ready" | "running" | "closed" | "stopped" | "error";
+  readonly updatedAt: string;
+}
+
+export interface ProviderStatusCacheShape {
+  readonly get: (threadId: ThreadId) => Effect.Effect<ProviderStatusEntry | undefined>;
+  readonly set: (threadId: ThreadId, entry: ProviderStatusEntry) => Effect.Effect<void>;
+  readonly delete: (threadId: ThreadId) => Effect.Effect<void>;
+}
+
+export class ProviderStatusCache extends ServiceMap.Service<
+  ProviderStatusCache,
+  ProviderStatusCacheShape
+>()("t3/provider/Services/ProviderStatusCache") {}