authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-26 10:35:37-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log8efbb31725f8500e1aff76655b04a3fab7b064fe
tree9f7853eaf32211f7576ba49ca511caa208927bfe
parentb4c74723d1a3072551e38d1bd3b5098e3872a51d

use effect/Cache instead of ad-hoc map caches

4 files changed, 155 insertions(+), 151 deletions(-)

apps/server/src/git/Layers/GitCore.ts+67-54
......@@ -1,11 +1,19 @@
1import { Effect, FileSystem, Layer, Path } from "effect";
1import { Cache, Data, Duration, Effect, Exit, FileSystem, Layer, Path } from "effect";
22
33import { GitCommandError } from "../Errors.ts";
44import { GitService } from "../Services/GitService.ts";
55import { GitCore, type GitCoreShape } from "../Services/GitCore.ts";
66
7const STATUS_UPSTREAM_REFRESH_INTERVAL_MS = 15_000;
8const STATUS_UPSTREAM_REFRESH_TIMEOUT_MS = 5_000;
7const STATUS_UPSTREAM_REFRESH_INTERVAL = Duration.seconds(15);
8const STATUS_UPSTREAM_REFRESH_TIMEOUT = Duration.seconds(5);
9const STATUS_UPSTREAM_REFRESH_CACHE_CAPACITY = 2_048;
10
11class StatusUpstreamRefreshCacheKey extends Data.Class<{
12 cwd: string;
13 upstreamRef: string;
14 remoteName: string;
15 upstreamBranch: string;
16}> {}
917
1018interface ExecuteGitOptions {
1119 timeoutMs?: number | undefined;
......@@ -13,10 +21,6 @@ interface ExecuteGitOptions {
1321 fallbackErrorMessage?: string | undefined;
1422}
1523
16function trimStdout(value: string): string {
17 return value.trim();
18}
19
2024function parseBranchAb(value: string): { ahead: number; behind: number } {
2125 const match = value.match(/^\+(\d+)\s+-(\d+)$/);
2226 if (!match) return { ahead: 0, behind: 0 };
......@@ -96,7 +100,6 @@ const makeGitCore = Effect.gen(function* () {
96100 const git = yield* GitService;
97101 const fileSystem = yield* FileSystem.FileSystem;
98102 const path = yield* Path.Path;
99 const lastUpstreamRefreshAt = new Map<string, number>();
100103
101104 const executeGit = (
102105 operation: string,
......@@ -162,14 +165,13 @@ const makeGitCore = Effect.gen(function* () {
162165 GitCommandError
163166 > =>
164167 Effect.gen(function* () {
165 const upstreamRef = trimStdout(
166 yield* runGitStdout(
167 "GitCore.resolveCurrentUpstream",
168 cwd,
169 ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{upstream}"],
170 true,
171 ),
172 );
168 const upstreamRef = yield* runGitStdout(
169 "GitCore.resolveCurrentUpstream",
170 cwd,
171 ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{upstream}"],
172 true,
173 ).pipe(Effect.map((stdout) => stdout.trim()));
174
173175 if (upstreamRef.length === 0 || upstreamRef === "@{upstream}") {
174176 return null;
175177 }
......@@ -215,36 +217,45 @@ const makeGitCore = Effect.gen(function* () {
215217 ["fetch", "--quiet", "--no-tags", upstream.remoteName, refspec],
216218 {
217219 allowNonZeroExit: true,
218 timeoutMs: STATUS_UPSTREAM_REFRESH_TIMEOUT_MS,
220 timeoutMs: Duration.toMillis(STATUS_UPSTREAM_REFRESH_TIMEOUT),
219221 },
220222 ).pipe(Effect.asVoid);
221223 };
222224
225 const statusUpstreamRefreshCache = yield* Cache.makeWith({
226 capacity: STATUS_UPSTREAM_REFRESH_CACHE_CAPACITY,
227 lookup: (cacheKey: StatusUpstreamRefreshCacheKey) =>
228 Effect.gen(function* () {
229 yield* fetchUpstreamRefForStatus(cacheKey.cwd, {
230 upstreamRef: cacheKey.upstreamRef,
231 remoteName: cacheKey.remoteName,
232 upstreamBranch: cacheKey.upstreamBranch,
233 });
234 return true as const;
235 }),
236 // Keep successful refreshes warm; drop failures immediately so next request can retry.
237 timeToLive: (exit) => (Exit.isSuccess(exit) ? STATUS_UPSTREAM_REFRESH_INTERVAL : Duration.zero),
238 });
239
223240 const refreshStatusUpstreamIfStale = (cwd: string): Effect.Effect<void, GitCommandError> =>
224241 Effect.gen(function* () {
225242 const upstream = yield* resolveCurrentUpstream(cwd);
226 if (!upstream) {
227 return;
228 }
229
230 const cacheKey = `${cwd}\u0000${upstream.upstreamRef}`;
231 const now = Date.now();
232 const lastRefreshAt = lastUpstreamRefreshAt.get(cacheKey) ?? 0;
233 if (now - lastRefreshAt < STATUS_UPSTREAM_REFRESH_INTERVAL_MS) {
234 return;
235 }
236
237 // Record attempt time first to avoid repeated fetch spikes when status is polled often.
238 lastUpstreamRefreshAt.set(cacheKey, now);
239 yield* fetchUpstreamRefForStatus(cwd, upstream);
243 if (!upstream) return;
244 yield* Cache.get(
245 statusUpstreamRefreshCache,
246 new StatusUpstreamRefreshCacheKey({
247 cwd,
248 upstreamRef: upstream.upstreamRef,
249 remoteName: upstream.remoteName,
250 upstreamBranch: upstream.upstreamBranch,
251 }),
252 );
240253 });
241254
242255 const refreshCheckedOutBranchUpstream = (cwd: string): Effect.Effect<void, GitCommandError> =>
243256 Effect.gen(function* () {
244257 const upstream = yield* resolveCurrentUpstream(cwd);
245 if (!upstream) {
246 return;
247 }
258 if (!upstream) return;
248259 yield* fetchUpstreamRef(cwd, upstream);
249260 });
250261
......@@ -394,8 +405,8 @@ const makeGitCore = Effect.gen(function* () {
394405 "diff",
395406 "--cached",
396407 "--name-status",
397 ]);
398 if (trimStdout(stagedSummary).length === 0) {
408 ]).pipe(Effect.map((stdout) => stdout.trim()));
409 if (stagedSummary.length === 0) {
399410 return null;
400411 }
401412
......@@ -420,9 +431,11 @@ const makeGitCore = Effect.gen(function* () {
420431 args.push("-m", trimmedBody);
421432 }
422433 yield* runGit("GitCore.commit.commit", cwd, args);
423 const commitSha = trimStdout(
424 yield* runGitStdout("GitCore.commit.revParseHead", cwd, ["rev-parse", "HEAD"]),
425 );
434 const commitSha = yield* runGitStdout("GitCore.commit.revParseHead", cwd, [
435 "rev-parse",
436 "HEAD",
437 ]).pipe(Effect.map((stdout) => stdout.trim()));
438
426439 return { commitSha };
427440 });
428441
......@@ -491,21 +504,23 @@ const makeGitCore = Effect.gen(function* () {
491504 "Current branch has no upstream configured. Push with upstream first.",
492505 );
493506 }
494 const beforeSha = trimStdout(
495 yield* runGitStdout(
496 "GitCore.pullCurrentBranch.beforeSha",
497 cwd,
498 ["rev-parse", "HEAD"],
499 true,
500 ),
501 );
507 const beforeSha = yield* runGitStdout(
508 "GitCore.pullCurrentBranch.beforeSha",
509 cwd,
510 ["rev-parse", "HEAD"],
511 true,
512 ).pipe(Effect.map((stdout) => stdout.trim()));
502513 yield* executeGit("GitCore.pullCurrentBranch.pull", cwd, ["pull", "--ff-only"], {
503514 timeoutMs: 30_000,
504515 fallbackErrorMessage: "git pull failed",
505516 });
506 const afterSha = trimStdout(
507 yield* runGitStdout("GitCore.pullCurrentBranch.afterSha", cwd, ["rev-parse", "HEAD"], true),
508 );
517 const afterSha = yield* runGitStdout(
518 "GitCore.pullCurrentBranch.afterSha",
519 cwd,
520 ["rev-parse", "HEAD"],
521 true,
522 ).pipe(Effect.map((stdout) => stdout.trim()));
523
509524 const refreshed = yield* statusDetails(cwd);
510525 return {
511526 status: beforeSha.length > 0 && beforeSha === afterSha ? "skipped_up_to_date" : "pulled",
......@@ -540,10 +555,8 @@ const makeGitCore = Effect.gen(function* () {
540555
541556 const readConfigValue: GitCoreShape["readConfigValue"] = (cwd, key) =>
542557 runGitStdout("GitCore.readConfigValue", cwd, ["config", "--get", key], true).pipe(
543 Effect.map((stdout) => {
544 const value = trimStdout(stdout);
545 return value.length > 0 ? value : null;
546 }),
558 Effect.map((stdout) => stdout.trim()),
559 Effect.map((trimmed) => (trimmed.length > 0 ? trimmed : null)),
547560 );
548561
549562 const listBranches: GitCoreShape["listBranches"] = (input) =>
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+15-38
......@@ -8,7 +8,7 @@ import {
88 type ThreadId,
99 type TurnId,
1010} from "@t3tools/contracts";
11import { Effect, Layer, Queue, Stream } from "effect";
11import { Cache, Duration, Effect, Layer, Option, Queue, Stream } from "effect";
1212
1313import { ProviderService } from "../../provider/Services/ProviderService.ts";
1414import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
......@@ -76,46 +76,23 @@ const serverCommandId = (tag: string): CommandId =>
7676 CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`);
7777
7878const HANDLED_TURN_START_KEY_MAX = 10_000;
79const HANDLED_TURN_START_KEY_TTL_MS = 30 * 60 * 1_000;
80
81function pruneHandledTurnStartKeys(handledKeys: Map<string, number>, nowMs: number): void {
82 const staleBefore = nowMs - HANDLED_TURN_START_KEY_TTL_MS;
83 for (const [key, seenAt] of handledKeys) {
84 if (seenAt >= staleBefore) {
85 break;
86 }
87 handledKeys.delete(key);
88 }
89
90 while (handledKeys.size > HANDLED_TURN_START_KEY_MAX) {
91 const oldestKey = handledKeys.keys().next().value;
92 if (!oldestKey) {
93 break;
94 }
95 handledKeys.delete(oldestKey);
96 }
97}
98
99function hasHandledTurnStartRecently(
100 handledKeys: Map<string, number>,
101 key: string,
102 nowMs = Date.now(),
103): boolean {
104 const seenAt = handledKeys.get(key);
105 if (seenAt !== undefined && nowMs - seenAt <= HANDLED_TURN_START_KEY_TTL_MS) {
106 handledKeys.set(key, nowMs);
107 return true;
108 }
109
110 handledKeys.set(key, nowMs);
111 pruneHandledTurnStartKeys(handledKeys, nowMs);
112 return false;
113}
79const HANDLED_TURN_START_KEY_TTL = Duration.minutes(30);
11480
11581const make = Effect.gen(function* () {
11682 const orchestrationEngine = yield* OrchestrationEngineService;
11783 const providerService = yield* ProviderService;
118 const handledTurnStartKeys = new Map<string, number>();
84 const handledTurnStartKeys = yield* Cache.make<string, true>({
85 capacity: HANDLED_TURN_START_KEY_MAX,
86 timeToLive: HANDLED_TURN_START_KEY_TTL,
87 lookup: () => Effect.succeed(true),
88 });
89
90 const hasHandledTurnStartRecently = (key: string) =>
91 Cache.getOption(handledTurnStartKeys, key).pipe(
92 Effect.flatMap((cached) =>
93 Cache.set(handledTurnStartKeys, key, true).pipe(Effect.as(Option.isSome(cached))),
94 ),
95 );
11996
12097 const appendProviderFailureActivity = (input: {
12198 readonly threadId: ThreadId;
......@@ -246,7 +223,7 @@ const make = Effect.gen(function* () {
246223 event: Extract<ProviderIntentEvent, { type: "thread.turn-start-requested" }>,
247224 ) {
248225 const key = turnStartKeyForEvent(event);
249 if (hasHandledTurnStartRecently(handledTurnStartKeys, key)) {
226 if (yield* hasHandledTurnStartRecently(key)) {
250227 return;
251228 }
252229
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+66-42
......@@ -7,7 +7,7 @@ import {
77 type ProviderRuntimeEvent,
88 type ProviderSessionId,
99} from "@t3tools/contracts";
10import { Cause, Effect, Layer, Queue, Stream } from "effect";
10import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "effect";
1111
1212import { ProviderService } from "../../provider/Services/ProviderService.ts";
1313import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
......@@ -19,6 +19,8 @@ import {
1919const providerTurnKey = (sessionId: ProviderSessionId, turnId: TurnId) => `${sessionId}:${turnId}`;
2020const providerCommandId = (event: ProviderRuntimeEvent, tag: string): CommandId =>
2121 CommandId.makeUnsafe(`provider:${event.eventId}:${tag}:${crypto.randomUUID()}`);
22const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000;
23const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120);
2224
2325function toTurnId(value: string | undefined): TurnId | undefined {
2426 return value === undefined ? undefined : TurnId.makeUnsafe(value);
......@@ -128,54 +130,76 @@ const make = Effect.gen(function* () {
128130 const orchestrationEngine = yield* OrchestrationEngineService;
129131 const providerService = yield* ProviderService;
130132
131 const turnMessageIdsByTurnKey = new Map<string, Set<MessageId>>();
133 const turnMessageIdsByTurnKey = yield* Cache.make<string, Set<MessageId>>({
134 capacity: TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY,
135 timeToLive: TURN_MESSAGE_IDS_BY_TURN_TTL,
136 lookup: () => Effect.succeed(new Set<MessageId>()),
137 });
132138
133139 const rememberAssistantMessageId = (
134140 sessionId: ProviderSessionId,
135141 turnId: TurnId,
136142 messageId: MessageId,
137 ) => {
138 const key = providerTurnKey(sessionId, turnId);
139 const existingIds = turnMessageIdsByTurnKey.get(key);
140 if (existingIds) {
141 existingIds.add(messageId);
142 } else {
143 turnMessageIdsByTurnKey.set(key, new Set([messageId]));
144 }
145 };
143 ) =>
144 Cache.getOption(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId)).pipe(
145 Effect.flatMap((existingIds) =>
146 Cache.set(
147 turnMessageIdsByTurnKey,
148 providerTurnKey(sessionId, turnId),
149 Option.match(existingIds, {
150 onNone: () => new Set([messageId]),
151 onSome: (ids) => {
152 const nextIds = new Set(ids);
153 nextIds.add(messageId);
154 return nextIds;
155 },
156 }),
157 ),
158 ),
159 );
146160
147161 const forgetAssistantMessageId = (
148162 sessionId: ProviderSessionId,
149163 turnId: TurnId,
150164 messageId: MessageId,
151 ) => {
152 const key = providerTurnKey(sessionId, turnId);
153 const existingIds = turnMessageIdsByTurnKey.get(key);
154 if (!existingIds) {
155 return;
156 }
157 existingIds.delete(messageId);
158 if (existingIds.size === 0) {
159 turnMessageIdsByTurnKey.delete(key);
160 }
161 };
165 ) =>
166 Cache.getOption(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId)).pipe(
167 Effect.flatMap((existingIds) =>
168 Option.match(existingIds, {
169 onNone: () => Effect.void,
170 onSome: (ids) => {
171 const nextIds = new Set(ids);
172 nextIds.delete(messageId);
173 if (nextIds.size === 0) {
174 return Cache.invalidate(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId));
175 }
176 return Cache.set(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId), nextIds);
177 },
178 }),
179 ),
180 );
162181
163 const getAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) => {
164 return turnMessageIdsByTurnKey.get(providerTurnKey(sessionId, turnId)) ?? new Set<MessageId>();
165 };
182 const getAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) =>
183 Cache.getOption(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId)).pipe(
184 Effect.map((existingIds) =>
185 Option.getOrElse(existingIds, (): Set<MessageId> => new Set<MessageId>()),
186 ),
187 );
166188
167 const clearAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) => {
168 turnMessageIdsByTurnKey.delete(providerTurnKey(sessionId, turnId));
169 };
189 const clearAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) =>
190 Cache.invalidate(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId));
170191
171 const clearTurnStateForSession = (sessionId: ProviderSessionId) => {
172 const prefix = `${sessionId}:`;
173 for (const key of turnMessageIdsByTurnKey.keys()) {
174 if (key.startsWith(prefix)) {
175 turnMessageIdsByTurnKey.delete(key);
176 }
177 }
178 };
192 const clearTurnStateForSession = (sessionId: ProviderSessionId) =>
193 Effect.gen(function* () {
194 const prefix = `${sessionId}:`;
195 const turnKeys = Array.from(yield* Cache.keys(turnMessageIdsByTurnKey));
196 yield* Effect.forEach(
197 turnKeys,
198 (key) =>
199 key.startsWith(prefix) ? Cache.invalidate(turnMessageIdsByTurnKey, key) : Effect.void,
200 { concurrency: 1 },
201 ).pipe(Effect.asVoid);
202 });
179203
180204 const processEvent = (event: ProviderRuntimeEvent) =>
181205 Effect.gen(function* () {
......@@ -243,7 +267,7 @@ const make = Effect.gen(function* () {
243267 );
244268 const turnId = toTurnId(event.turnId);
245269 if (turnId) {
246 rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId);
270 yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId);
247271 }
248272
249273 yield* orchestrationEngine.dispatch({
......@@ -261,7 +285,7 @@ const make = Effect.gen(function* () {
261285 const assistantMessageId = MessageId.makeUnsafe(`assistant:${event.itemId}`);
262286 const turnId = toTurnId(event.turnId);
263287 if (turnId) {
264 rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId);
288 yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId);
265289 }
266290
267291 yield* orchestrationEngine.dispatch({
......@@ -274,14 +298,14 @@ const make = Effect.gen(function* () {
274298 });
275299
276300 if (turnId) {
277 forgetAssistantMessageId(event.sessionId, turnId, assistantMessageId);
301 yield* forgetAssistantMessageId(event.sessionId, turnId, assistantMessageId);
278302 }
279303 }
280304
281305 if (event.type === "turn.completed") {
282306 const turnId = toTurnId(event.turnId);
283307 if (turnId) {
284 const assistantMessageIds = getAssistantMessageIdsForTurn(event.sessionId, turnId);
308 const assistantMessageIds = yield* getAssistantMessageIdsForTurn(event.sessionId, turnId);
285309 yield* Effect.forEach(assistantMessageIds, (assistantMessageId) =>
286310 orchestrationEngine.dispatch({
287311 type: "thread.message.assistant.complete",
......@@ -292,12 +316,12 @@ const make = Effect.gen(function* () {
292316 createdAt: now,
293317 }),
294318 ).pipe(Effect.asVoid);
295 clearAssistantMessageIdsForTurn(event.sessionId, turnId);
319 yield* clearAssistantMessageIdsForTurn(event.sessionId, turnId);
296320 }
297321 }
298322
299323 if (event.type === "session.exited") {
300 clearTurnStateForSession(event.sessionId);
324 yield* clearTurnStateForSession(event.sessionId);
301325 }
302326
303327 if (event.type === "runtime.error") {
apps/server/src/provider/Layers/ProviderService.ts+7-17
......@@ -12,7 +12,6 @@
1212import {
1313 NonNegativeInt,
1414 ProviderSessionId,
15 ProviderThreadId,
1615 ThreadId,
1716 ProviderInterruptTurnInput,
1817 ProviderRespondToRequestInput,
......@@ -94,14 +93,6 @@ function toRuntimePayloadFromSession(session: ProviderSession): Record<string, u
9493 };
9594}
9695
97function asProviderSessionId(value: string | ProviderSessionId): ProviderSessionId {
98 return typeof value === "string" ? ProviderSessionId.makeUnsafe(value) : value;
99}
100
101function asProviderThreadId(value: string | ProviderThreadId): ProviderThreadId {
102 return typeof value === "string" ? ProviderThreadId.makeUnsafe(value) : value;
103}
104
10596const makeProviderService = (options?: ProviderServiceLiveOptions) =>
10697 Effect.gen(function* () {
10798 const canonicalEventLogger =
......@@ -126,7 +117,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
126117 if (liveSessionId === event.sessionId) {
127118 return {
128119 ...event,
129 sessionId: ProviderSessionId.makeUnsafe(staleSessionId),
120 sessionId: staleSessionId,
130121 } satisfies ProviderRuntimeEvent;
131122 }
132123 }
......@@ -162,9 +153,8 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
162153 );
163154 }
164155
165 const providerSessionId = asProviderSessionId(session.sessionId);
166156 yield* directory.upsert({
167 sessionId: providerSessionId,
157 sessionId: session.sessionId,
168158 provider: session.provider,
169159 threadId,
170160 providerThreadId,
......@@ -444,7 +434,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
444434 sessionId: routed.sessionId,
445435 });
446436 const threadId = yield* directory
447 .getThreadId(asProviderSessionId(input.sessionId))
437 .getThreadId(input.sessionId)
448438 .pipe(Effect.map(Option.getOrUndefined));
449439 if (!threadId) {
450440 return yield* toValidationError(
......@@ -453,10 +443,10 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
453443 );
454444 }
455445 yield* directory.upsert({
456 sessionId: asProviderSessionId(input.sessionId),
446 sessionId: input.sessionId,
457447 provider: routed.adapter.provider,
458448 threadId,
459 providerThreadId: asProviderThreadId(turn.threadId),
449 providerThreadId: turn.threadId,
460450 status: "running",
461451 ...(turn.resumeCursor !== undefined ? { resumeCursor: turn.resumeCursor } : {}),
462452 runtimePayload: {
......@@ -514,10 +504,10 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
514504 yield* routed.adapter.stopSession(routed.sessionId);
515505 }
516506 if (routed.sessionId !== input.sessionId) {
517 yield* directory.remove(asProviderSessionId(routed.sessionId));
507 yield* directory.remove(routed.sessionId);
518508 yield* clearAliasesReferencing(routed.sessionId);
519509 }
520 yield* directory.remove(asProviderSessionId(input.sessionId));
510 yield* directory.remove(input.sessionId);
521511 yield* clearAliasesReferencing(input.sessionId);
522512 });
523513