authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-06-19 21:18:55-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-19 21:18:55-07:00
log01492ebd3b624e544e6375c12540b741c65256b9
treef9b802a8d8d83645996844a52fd21180763b21de
parent93fd9721086d4325794830a29a7df0592d120683
signature Signed by PGP key B5690EEEBB952194

[codex] Refactor agent awareness relay service (#3197)

Co-authored-by: codex <codex@users.noreply.github.com>

1 files changed, 23 insertions(+), 25 deletions(-)

apps/server/src/relay/AgentAwarenessRelay.ts+23-25
......@@ -1,8 +1,3 @@
1import {
2 RelayApi,
3 type RelayAgentActivityPublishProofPayload,
4 type RelayAgentActivityState,
5} from "@t3tools/contracts/relay";
61import type {
72 EnvironmentId,
83 OrchestrationEvent,
......@@ -10,13 +5,18 @@ import type {
105 OrchestrationThreadShell,
116 ThreadId,
127} from "@t3tools/contracts";
8import {
9 RelayApi,
10 type RelayAgentActivityPublishProofPayload,
11 type RelayAgentActivityState,
12} from "@t3tools/contracts/relay";
1313import { projectThreadAwareness } from "@t3tools/shared/agentAwareness";
1414import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
1515import { withRelayClientTracing } from "@t3tools/shared/relayTracing";
1616import {
17 normalizeRelayIssuer,
1718 RELAY_ACTIVITY_PUBLISH_TYP,
1819 signRelayJwt,
19 normalizeRelayIssuer,
2020} from "@t3tools/shared/relayJwt";
2121import * as Cause from "effect/Cause";
2222import * as Context from "effect/Context";
......@@ -28,31 +28,29 @@ import * as Option from "effect/Option";
2828import * as Ref from "effect/Ref";
2929import type * as Scope from "effect/Scope";
3030import * as Stream from "effect/Stream";
31import { FetchHttpClient } from "effect/unstable/http";
31import * as FetchHttpClient from "effect/unstable/http/FetchHttpClient";
3232import * as HttpClient from "effect/unstable/http/HttpClient";
3333import * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";
3434import * as HttpApiClient from "effect/unstable/httpapi/HttpApiClient";
3535
3636import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
37import { getOrCreateEnvironmentKeyPairFromSecretStore } from "../cloud/environmentKeys.ts";
3837import {
3938 PUBLISH_AGENT_ACTIVITY_SECRET,
4039 RELAY_ENVIRONMENT_CREDENTIAL_SECRET,
4140 RELAY_ISSUER_SECRET,
4241 RELAY_URL_SECRET,
4342} from "../cloud/config.ts";
44import { ServerEnvironment } from "../environment/Services/ServerEnvironment.ts";
45import { OrchestrationEngineService } from "../orchestration/Services/OrchestrationEngine.ts";
46import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts";
47
48export interface AgentAwarenessRelayShape {
49 readonly publishThread: (threadId: ThreadId) => Effect.Effect<void>;
50 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
51}
43import { getOrCreateEnvironmentKeyPairFromSecretStore } from "../cloud/environmentKeys.ts";
44import * as ServerEnvironment from "../environment/Services/ServerEnvironment.ts";
45import * as OrchestrationEngine from "../orchestration/Services/OrchestrationEngine.ts";
46import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSnapshotQuery.ts";
5247
5348export class AgentAwarenessRelay extends Context.Service<
5449 AgentAwarenessRelay,
55 AgentAwarenessRelayShape
50 {
51 readonly publishThread: (threadId: ThreadId) => Effect.Effect<void>;
52 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
53 }
5654>()("t3/relay/AgentAwarenessRelay") {}
5755
5856export function eventThreadId(event: OrchestrationEvent): ThreadId | null {
......@@ -265,11 +263,11 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: {
265263 .map((thread) => thread.id);
266264}
267265
268const make = Effect.gen(function* () {
266export const make = Effect.gen(function* () {
269267 const secrets = yield* ServerSecretStore.ServerSecretStore;
270 const serverEnvironment = yield* ServerEnvironment;
271 const snapshotQuery = yield* ProjectionSnapshotQuery;
272 const orchestrationEngine = yield* OrchestrationEngineService;
268 const serverEnvironment = yield* ServerEnvironment.ServerEnvironment;
269 const snapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
270 const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService;
273271 const crypto = yield* Crypto.Crypto;
274272 const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets);
275273 const activeSnapshotPublishedRef = yield* Ref.make(false);
......@@ -417,7 +415,7 @@ const make = Effect.gen(function* () {
417415 });
418416 });
419417
420 const publishThread: AgentAwarenessRelayShape["publishThread"] = (threadId) =>
418 const publishThread: AgentAwarenessRelay["Service"]["publishThread"] = (threadId) =>
421419 publishThreadUnsafe(threadId).pipe(
422420 Effect.catchCause((cause) => {
423421 return Effect.logWarning("agent activity publish failed", {
......@@ -480,7 +478,7 @@ const make = Effect.gen(function* () {
480478
481479 const worker = yield* makeDrainableWorker(publishThread);
482480
483 const start: AgentAwarenessRelayShape["start"] = Effect.fn("AgentAwarenessRelay.start")(
481 const start: AgentAwarenessRelay["Service"]["start"] = Effect.fn("AgentAwarenessRelay.start")(
484482 function* () {
485483 const [relayConfig, publishEnabled] = yield* Effect.all([
486484 readRelayConfig.pipe(Effect.orElseSucceed(() => null)),
......@@ -536,10 +534,10 @@ const make = Effect.gen(function* () {
536534 },
537535 );
538536
539 return {
537 return AgentAwarenessRelay.of({
540538 publishThread,
541539 start,
542 } satisfies AgentAwarenessRelayShape;
540 });
543541});
544542
545543export const layer = Layer.effect(AgentAwarenessRelay, make);