History

Prepare datamodel for multi-environment (#1765)

Julius Marminge committed 7 months ago

Commit b96308f1 parent 72b7f90121 files · +10206 −4969Browse files

Showing the changes to apps/server/src (20 of 121 files). Show all

Unified · Split

20 files
  1. apps/server/src/config.ts +2 −0
  2. apps/server/src/environment/Layers/ServerEnvironment.test.ts +124 −0
  3. apps/server/src/environment/Layers/ServerEnvironment.ts +94 −0
  4. apps/server/src/environment/Services/ServerEnvironment.ts +13 −0
  5. apps/server/src/git/Layers/GitManager.test.ts +3 −3
  6. apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +2 −0
  7. apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +5 −0
  8. apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +2 −0
  9. apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +6 −1
  10. apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +278 −256
  11. apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +2 −0
  12. apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +2 −0
  13. apps/server/src/project/Layers/RepositoryIdentityResolver.test.ts +193 −0
  14. apps/server/src/project/Layers/RepositoryIdentityResolver.ts +147 −0
  15. apps/server/src/project/Services/RepositoryIdentityResolver.ts +12 −0
  16. apps/server/src/server.test.ts +245 −1
  17. apps/server/src/server.ts +4 −0
  18. apps/server/src/serverLifecycleEvents.test.ts +10 −0
  19. apps/server/src/serverRuntimeStartup.ts +12 −2
  20. apps/server/src/ws.ts +55 −1
apps/server/src/config.ts+2−0
@@ -30,6 +30,7 @@ export interface ServerDerivedPaths { · 29 unmodified lines
30 readonly providerEventLogPath: string;30 readonly providerEventLogPath: string;
31 readonly terminalLogsDir: string;31 readonly terminalLogsDir: string;
32 readonly anonymousIdPath: string;32 readonly anonymousIdPath: string;
33 readonly environmentIdPath: string;
33}34}
34 35
35/**36/**
@@ -83,6 +84,7 @@ export const deriveServerPaths = Effect.fn(function* ( · 47 unmodified lines
83 providerEventLogPath: join(providerLogsDir, "events.log"),84 providerEventLogPath: join(providerLogsDir, "events.log"),
84 terminalLogsDir: join(logsDir, "terminals"),85 terminalLogsDir: join(logsDir, "terminals"),
85 anonymousIdPath: join(stateDir, "anonymous-id"),86 anonymousIdPath: join(stateDir, "anonymous-id"),
87 environmentIdPath: join(stateDir, "environment-id"),
86 };88 };
87});89});
88 90
apps/server/src/environment/Layers/ServerEnvironment.test.ts+124−0
@@ -0,0 +1,124 @@
1import * as nodePath from "node:path";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { expect, it } from "@effect/vitest";
4import { Effect, Exit, FileSystem, Layer, PlatformError } from "effect";
5
6import { deriveServerPaths, ServerConfig, type ServerConfigShape } from "../../config.ts";
7import { ServerEnvironment } from "../Services/ServerEnvironment.ts";
8import { ServerEnvironmentLive } from "./ServerEnvironment.ts";
9
10const makeServerEnvironmentLayer = (baseDir: string) =>
11 ServerEnvironmentLive.pipe(Layer.provide(ServerConfig.layerTest(process.cwd(), baseDir)));
12
13const makeServerConfig = Effect.fn(function* (baseDir: string) {
14 const derivedPaths = yield* deriveServerPaths(baseDir, undefined);
15
16 return {
17 ...derivedPaths,
18 logLevel: "Error",
19 traceMinLevel: "Info",
20 traceTimingEnabled: true,
21 traceBatchWindowMs: 200,
22 traceMaxBytes: 10 * 1024 * 1024,
23 traceMaxFiles: 10,
24 otlpTracesUrl: undefined,
25 otlpMetricsUrl: undefined,
26 otlpExportIntervalMs: 10_000,
27 otlpServiceName: "t3-server",
28 cwd: process.cwd(),
29 baseDir,
30 mode: "web",
31 autoBootstrapProjectFromCwd: false,
32 logWebSocketEvents: false,
33 port: 0,
34 host: undefined,
35 authToken: undefined,
36 staticDir: undefined,
37 devUrl: undefined,
38 noBrowser: false,
39 } satisfies ServerConfigShape;
40});
41
42it.layer(NodeServices.layer)("ServerEnvironmentLive", (it) => {
43 it.effect("persists the environment id across service restarts", () =>
44 Effect.gen(function* () {
45 const fileSystem = yield* FileSystem.FileSystem;
46 const baseDir = yield* fileSystem.makeTempDirectoryScoped({
47 prefix: "t3-server-environment-test-",
48 });
49
50 const first = yield* Effect.gen(function* () {
51 const serverEnvironment = yield* ServerEnvironment;
52 return yield* serverEnvironment.getDescriptor;
53 }).pipe(Effect.provide(makeServerEnvironmentLayer(baseDir)));
54 const second = yield* Effect.gen(function* () {
55 const serverEnvironment = yield* ServerEnvironment;
56 return yield* serverEnvironment.getDescriptor;
57 }).pipe(Effect.provide(makeServerEnvironmentLayer(baseDir)));
58
59 expect(first.environmentId).toBe(second.environmentId);
60 expect(second.capabilities.repositoryIdentity).toBe(true);
61 }),
62 );
63
64 it.effect("fails instead of overwriting a persisted id when reading the file errors", () =>
65 Effect.gen(function* () {
66 const fileSystem = yield* FileSystem.FileSystem;
67 const baseDir = yield* fileSystem.makeTempDirectoryScoped({
68 prefix: "t3-server-environment-read-error-test-",
69 });
70 const serverConfig = yield* makeServerConfig(baseDir);
71 const environmentIdPath = serverConfig.environmentIdPath;
72 yield* fileSystem.makeDirectory(nodePath.dirname(environmentIdPath), { recursive: true });
73 yield* fileSystem.writeFileString(environmentIdPath, "persisted-environment-id\n");
74 const writeAttempts: string[] = [];
75 const failingFileSystemLayer = FileSystem.layerNoop({
76 exists: (path) => Effect.succeed(path === environmentIdPath),
77 readFileString: (path) =>
78 path === environmentIdPath
79 ? Effect.fail(
80 PlatformError.systemError({
81 _tag: "PermissionDenied",
82 module: "FileSystem",
83 method: "readFileString",
84 description: "permission denied",
85 pathOrDescriptor: path,
86 }),
87 )
88 : Effect.fail(
89 PlatformError.systemError({
90 _tag: "NotFound",
91 module: "FileSystem",
92 method: "readFileString",
93 description: "not found",
94 pathOrDescriptor: path,
95 }),
96 ),
97 writeFileString: (path) => {
98 writeAttempts.push(path);
99 return Effect.void;
100 },
101 });
102
103 const exit = yield* Effect.gen(function* () {
104 const serverEnvironment = yield* ServerEnvironment;
105 return yield* serverEnvironment.getDescriptor;
106 }).pipe(
107 Effect.provide(
108 ServerEnvironmentLive.pipe(
109 Layer.provide(
110 Layer.merge(Layer.succeed(ServerConfig, serverConfig), failingFileSystemLayer),
111 ),
112 ),
113 ),
114 Effect.exit,
115 );
116
117 expect(Exit.isFailure(exit)).toBe(true);
118 expect(writeAttempts).toEqual([]);
119 expect(yield* fileSystem.readFileString(environmentIdPath)).toBe(
120 "persisted-environment-id\n",
121 );
122 }),
123 );
124});
apps/server/src/environment/Layers/ServerEnvironment.ts+94−0
@@ -0,0 +1,94 @@
1import { EnvironmentId, type ExecutionEnvironmentDescriptor } from "@t3tools/contracts";
2import { Effect, FileSystem, Layer, Path, Random } from "effect";
3
4import { ServerConfig } from "../../config.ts";
5import { ServerEnvironment, type ServerEnvironmentShape } from "../Services/ServerEnvironment.ts";
6import { version } from "../../../package.json" with { type: "json" };
7
8function platformOs(): ExecutionEnvironmentDescriptor["platform"]["os"] {
9 switch (process.platform) {
10 case "darwin":
11 return "darwin";
12 case "linux":
13 return "linux";
14 case "win32":
15 return "windows";
16 default:
17 return "unknown";
18 }
19}
20
21function platformArch(): ExecutionEnvironmentDescriptor["platform"]["arch"] {
22 switch (process.arch) {
23 case "arm64":
24 return "arm64";
25 case "x64":
26 return "x64";
27 default:
28 return "other";
29 }
30}
31
32export const makeServerEnvironment = Effect.fn("makeServerEnvironment")(function* () {
33 const fileSystem = yield* FileSystem.FileSystem;
34 const path = yield* Path.Path;
35 const serverConfig = yield* ServerConfig;
36
37 const readPersistedEnvironmentId = Effect.gen(function* () {
38 const exists = yield* fileSystem
39 .exists(serverConfig.environmentIdPath)
40 .pipe(Effect.orElseSucceed(() => false));
41 if (!exists) {
42 return null;
43 }
44
45 const raw = yield* fileSystem
46 .readFileString(serverConfig.environmentIdPath)
47 .pipe(Effect.map((value) => value.trim()));
48
49 return raw.length > 0 ? raw : null;
50 });
51
52 const persistEnvironmentId = (value: string) =>
53 fileSystem.writeFileString(serverConfig.environmentIdPath, `${value}\n`);
54
55 const environmentIdRaw = yield* Effect.gen(function* () {
56 const persisted = yield* readPersistedEnvironmentId;
57 if (persisted) {
58 return persisted;
59 }
60
61 const generated = yield* Random.nextUUIDv4;
62 yield* persistEnvironmentId(generated);
63 return generated;
64 });
65
66 const environmentId = EnvironmentId.makeUnsafe(environmentIdRaw);
67 const cwdBaseName = path.basename(serverConfig.cwd).trim();
68 const label =
69 serverConfig.mode === "desktop"
70 ? "Local environment"
71 : cwdBaseName.length > 0
72 ? cwdBaseName
73 : "T3 environment";
74
75 const descriptor: ExecutionEnvironmentDescriptor = {
76 environmentId,
77 label,
78 platform: {
79 os: platformOs(),
80 arch: platformArch(),
81 },
82 serverVersion: version,
83 capabilities: {
84 repositoryIdentity: true,
85 },
86 };
87
88 return {
89 getEnvironmentId: Effect.succeed(environmentId),
90 getDescriptor: Effect.succeed(descriptor),
91 } satisfies ServerEnvironmentShape;
92});
93
94export const ServerEnvironmentLive = Layer.effect(ServerEnvironment, makeServerEnvironment());
apps/server/src/environment/Services/ServerEnvironment.ts+13−0
@@ -0,0 +1,13 @@
1import type { EnvironmentId, ExecutionEnvironmentDescriptor } from "@t3tools/contracts";
2import { ServiceMap } from "effect";
3import type { Effect } from "effect";
4
5export interface ServerEnvironmentShape {
6 readonly getEnvironmentId: Effect.Effect<EnvironmentId>;
7 readonly getDescriptor: Effect.Effect<ExecutionEnvironmentDescriptor>;
8}
9
10export class ServerEnvironment extends ServiceMap.Service<
11 ServerEnvironment,
12 ServerEnvironmentShape
13>()("t3/environment/Services/ServerEnvironment") {}
apps/server/src/git/Layers/GitManager.test.ts+3−3
@@ -854,7 +854,7 @@ it.layer(GitManagerTestLayer)("GitManager", (it) => { · 853 unmodified lines
854 "pr list --head jasonLaster:statemachine --state all --limit 20 --json number,title,url,baseRefName,headRefName,state,mergedAt,updatedAt,isCrossRepository,headRepository,headRepositoryOwner",854 "pr list --head jasonLaster:statemachine --state all --limit 20 --json number,title,url,baseRefName,headRefName,state,mergedAt,updatedAt,isCrossRepository,headRepository,headRepositoryOwner",
855 );855 );
856 }),856 }),
857 12_000,857 20_000,
858 );858 );
859 859
860 it.effect(860 it.effect(
@@ -962,7 +962,7 @@ it.layer(GitManagerTestLayer)("GitManager", (it) => { · 101 unmodified lines
962 ),962 ),
963 ).toBe(false);963 ).toBe(false);
964 }),964 }),
965 12_000,965 20_000,
966 );966 );
967 967
968 it.effect("status returns merged PR state when latest PR was merged", () =>968 it.effect("status returns merged PR state when latest PR was merged", () =>
@@ -1685,7 +1685,7 @@ it.layer(GitManagerTestLayer)("GitManager", (it) => { · 716 unmodified lines
1685 false,1685 false,
1686 );1686 );
1687 }),1687 }),
1688 12_000,1688 20_000,
1689 );1689 );
1690 1690
1691 it.effect(1691 it.effect(
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+2−0
@@ -20,6 +20,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; · 19 unmodified lines
20import { CheckpointStoreLive } from "../../checkpointing/Layers/CheckpointStore.ts";20import { CheckpointStoreLive } from "../../checkpointing/Layers/CheckpointStore.ts";
21import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts";21import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts";
22import { GitCoreLive } from "../../git/Layers/GitCore.ts";22import { GitCoreLive } from "../../git/Layers/GitCore.ts";
23import { RepositoryIdentityResolverLive } from "../../project/Layers/RepositoryIdentityResolver.ts";
23import { CheckpointReactorLive } from "./CheckpointReactor.ts";24import { CheckpointReactorLive } from "./CheckpointReactor.ts";
24import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";25import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
25import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";26import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
@@ -256,6 +257,7 @@ describe("CheckpointReactor", () => { · 230 unmodified lines
256 Layer.provide(OrchestrationProjectionPipelineLive),257 Layer.provide(OrchestrationProjectionPipelineLive),
257 Layer.provide(OrchestrationEventStoreLive),258 Layer.provide(OrchestrationEventStoreLive),
258 Layer.provide(OrchestrationCommandReceiptRepositoryLive),259 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
260 Layer.provide(RepositoryIdentityResolverLive),
259 Layer.provide(SqlitePersistenceMemory),261 Layer.provide(SqlitePersistenceMemory),
260 );262 );
261 263
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+5−0
@@ -19,6 +19,7 @@ import { · 18 unmodified lines
19 OrchestrationEventStore,19 OrchestrationEventStore,
20 type OrchestrationEventStoreShape,20 type OrchestrationEventStoreShape,
21} from "../../persistence/Services/OrchestrationEventStore.ts";21} from "../../persistence/Services/OrchestrationEventStore.ts";
22import { RepositoryIdentityResolverLive } from "../../project/Layers/RepositoryIdentityResolver.ts";
22import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";23import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
23import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";24import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
24import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";25import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
@@ -45,6 +46,7 @@ async function createOrchestrationSystem() { · 20 unmodified lines
45 Layer.provide(OrchestrationProjectionPipelineLive),46 Layer.provide(OrchestrationProjectionPipelineLive),
46 Layer.provide(OrchestrationEventStoreLive),47 Layer.provide(OrchestrationEventStoreLive),
47 Layer.provide(OrchestrationCommandReceiptRepositoryLive),48 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
49 Layer.provide(RepositoryIdentityResolverLive),
48 Layer.provide(SqlitePersistenceMemory),50 Layer.provide(SqlitePersistenceMemory),
49 Layer.provideMerge(ServerConfigLayer),51 Layer.provideMerge(ServerConfigLayer),
50 Layer.provideMerge(NodeServices.layer),52 Layer.provideMerge(NodeServices.layer),
@@ -623,6 +625,7 @@ describe("OrchestrationEngine", () => { · 572 unmodified lines
623 Layer.provide(OrchestrationProjectionPipelineLive),625 Layer.provide(OrchestrationProjectionPipelineLive),
624 Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)),626 Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)),
625 Layer.provide(OrchestrationCommandReceiptRepositoryLive),627 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
628 Layer.provide(RepositoryIdentityResolverLive),
626 Layer.provide(SqlitePersistenceMemory),629 Layer.provide(SqlitePersistenceMemory),
627 Layer.provideMerge(ServerConfigLayer),630 Layer.provideMerge(ServerConfigLayer),
628 Layer.provideMerge(NodeServices.layer),631 Layer.provideMerge(NodeServices.layer),
@@ -719,6 +722,7 @@ describe("OrchestrationEngine", () => { · 90 unmodified lines
719 Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)),722 Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)),
720 Layer.provide(OrchestrationEventStoreLive),723 Layer.provide(OrchestrationEventStoreLive),
721 Layer.provide(OrchestrationCommandReceiptRepositoryLive),724 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
725 Layer.provide(RepositoryIdentityResolverLive),
722 Layer.provide(SqlitePersistenceMemory),726 Layer.provide(SqlitePersistenceMemory),
723 ),727 ),
724 );728 );
@@ -861,6 +865,7 @@ describe("OrchestrationEngine", () => { · 136 unmodified lines
861 Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)),865 Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)),
862 Layer.provide(Layer.succeed(OrchestrationEventStore, nonTransactionalStore)),866 Layer.provide(Layer.succeed(OrchestrationEventStore, nonTransactionalStore)),
863 Layer.provide(OrchestrationCommandReceiptRepositoryLive),867 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
868 Layer.provide(RepositoryIdentityResolverLive),
864 Layer.provide(SqlitePersistenceMemory),869 Layer.provide(SqlitePersistenceMemory),
865 ),870 ),
866 );871 );
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+2−0
@@ -20,6 +20,7 @@ import { · 19 unmodified lines
20 SqlitePersistenceMemory,20 SqlitePersistenceMemory,
21} from "../../persistence/Layers/Sqlite.ts";21} from "../../persistence/Layers/Sqlite.ts";
22import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";22import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
23import { RepositoryIdentityResolverLive } from "../../project/Layers/RepositoryIdentityResolver.ts";
23import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";24import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
24import {25import {
25 ORCHESTRATION_PROJECTOR_NAMES,26 ORCHESTRATION_PROJECTOR_NAMES,
@@ -1846,6 +1847,7 @@ const engineLayer = it.layer( · 1820 unmodified lines
1846 Layer.provide(OrchestrationProjectionPipelineLive),1847 Layer.provide(OrchestrationProjectionPipelineLive),
1847 Layer.provide(OrchestrationEventStoreLive),1848 Layer.provide(OrchestrationEventStoreLive),
1848 Layer.provide(OrchestrationCommandReceiptRepositoryLive),1849 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
1850 Layer.provide(RepositoryIdentityResolverLive),
1849 Layer.provideMerge(SqlitePersistenceMemory),1851 Layer.provideMerge(SqlitePersistenceMemory),
1850 Layer.provideMerge(1852 Layer.provideMerge(
1851 ServerConfig.layerTest(process.cwd(), {1853 ServerConfig.layerTest(process.cwd(), {
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+6−1
@@ -4,6 +4,7 @@ import { Effect, Layer } from "effect"; · 3 unmodified lines
4import * as SqlClient from "effect/unstable/sql/SqlClient";4import * as SqlClient from "effect/unstable/sql/SqlClient";
5 5
6import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";6import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
7import { RepositoryIdentityResolverLive } from "../../project/Layers/RepositoryIdentityResolver.ts";
7import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";8import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
8import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";9import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
9import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";10import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
@@ -15,7 +16,10 @@ const asEventId = (value: string): EventId => EventId.makeUnsafe(value); · 5 unmodified lines
15const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value);16const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.makeUnsafe(value);
16 17
17const projectionSnapshotLayer = it.layer(18const projectionSnapshotLayer = it.layer(
18 OrchestrationProjectionSnapshotQueryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),19 OrchestrationProjectionSnapshotQueryLive.pipe(
20 Layer.provideMerge(RepositoryIdentityResolverLive),
21 Layer.provideMerge(SqlitePersistenceMemory),
22 ),
19);23);
20 24
21projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {25projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
@@ -234,6 +238,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { · 212 unmodified lines
234 id: asProjectId("project-1"),238 id: asProjectId("project-1"),
235 title: "Project 1",239 title: "Project 1",
236 workspaceRoot: "/tmp/project-1",240 workspaceRoot: "/tmp/project-1",
241 repositoryIdentity: null,
237 defaultModelSelection: {242 defaultModelSelection: {
238 provider: "codex",243 provider: "codex",
239 model: "gpt-5-codex",244 model: "gpt-5-codex",
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+278−256
@@ -38,6 +38,7 @@ import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionTh · 37 unmodified lines
38import { ProjectionThreadProposedPlan } from "../../persistence/Services/ProjectionThreadProposedPlans.ts";38import { ProjectionThreadProposedPlan } from "../../persistence/Services/ProjectionThreadProposedPlans.ts";
39import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts";39import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts";
40import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts";40import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts";
41import { RepositoryIdentityResolver } from "../../project/Services/RepositoryIdentityResolver.ts";
41import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";42import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
42import {43import {
43 ProjectionSnapshotQuery,44 ProjectionSnapshotQuery,
@@ -163,6 +164,8 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st · 119 unmodified lines
163 164
164const makeProjectionSnapshotQuery = Effect.gen(function* () {165const makeProjectionSnapshotQuery = Effect.gen(function* () {
165 const sql = yield* SqlClient.SqlClient;166 const sql = yield* SqlClient.SqlClient;
167 const repositoryIdentityResolver = yield* RepositoryIdentityResolver;
168 const repositoryIdentityResolutionConcurrency = 4;
166 169
167 const listProjectRows = SqlSchema.findAll({170 const listProjectRows = SqlSchema.findAll({
168 Request: Schema.Void,171 Request: Schema.Void,
@@ -436,269 +439,283 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 267 unmodified lines
436 const getSnapshot: ProjectionSnapshotQueryShape["getSnapshot"] = () =>439 const getSnapshot: ProjectionSnapshotQueryShape["getSnapshot"] = () =>
437 sql440 sql
438 .withTransaction(441 .withTransaction(
439 Effect.gen(function* () {442 Effect.all([
440 const [443 listProjectRows(undefined).pipe(
441 projectRows,444 Effect.mapError(
442 threadRows,445 toPersistenceSqlOrDecodeError(
443 messageRows,446 "ProjectionSnapshotQuery.getSnapshot:listProjects:query",
444 proposedPlanRows,447 "ProjectionSnapshotQuery.getSnapshot:listProjects:decodeRows",
445 activityRows,
446 sessionRows,
447 checkpointRows,
448 latestTurnRows,
449 stateRows,
450 ] = yield* Effect.all([
451 listProjectRows(undefined).pipe(
452 Effect.mapError(
453 toPersistenceSqlOrDecodeError(
454 "ProjectionSnapshotQuery.getSnapshot:listProjects:query",
455 "ProjectionSnapshotQuery.getSnapshot:listProjects:decodeRows",
456 ),
457 ),
458 ),
459 listThreadRows(undefined).pipe(
460 Effect.mapError(
461 toPersistenceSqlOrDecodeError(
462 "ProjectionSnapshotQuery.getSnapshot:listThreads:query",
463 "ProjectionSnapshotQuery.getSnapshot:listThreads:decodeRows",
464 ),
465 ),448 ),
466 ),449 ),
467 listThreadMessageRows(undefined).pipe(450 ),
468 Effect.mapError(451 listThreadRows(undefined).pipe(
469 toPersistenceSqlOrDecodeError(452 Effect.mapError(
470 "ProjectionSnapshotQuery.getSnapshot:listThreadMessages:query",453 toPersistenceSqlOrDecodeError(
471 "ProjectionSnapshotQuery.getSnapshot:listThreadMessages:decodeRows",454 "ProjectionSnapshotQuery.getSnapshot:listThreads:query",
472 ),455 "ProjectionSnapshotQuery.getSnapshot:listThreads:decodeRows",
473 ),456 ),
474 ),457 ),
475 listThreadProposedPlanRows(undefined).pipe(458 ),
476 Effect.mapError(459 listThreadMessageRows(undefined).pipe(
477 toPersistenceSqlOrDecodeError(460 Effect.mapError(
478 "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:query",461 toPersistenceSqlOrDecodeError(
479 "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:decodeRows",462 "ProjectionSnapshotQuery.getSnapshot:listThreadMessages:query",
480 ),463 "ProjectionSnapshotQuery.getSnapshot:listThreadMessages:decodeRows",
481 ),464 ),
482 ),465 ),
483 listThreadActivityRows(undefined).pipe(466 ),
484 Effect.mapError(467 listThreadProposedPlanRows(undefined).pipe(
485 toPersistenceSqlOrDecodeError(468 Effect.mapError(
486 "ProjectionSnapshotQuery.getSnapshot:listThreadActivities:query",469 toPersistenceSqlOrDecodeError(
487 "ProjectionSnapshotQuery.getSnapshot:listThreadActivities:decodeRows",470 "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:query",
488 ),471 "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:decodeRows",
489 ),472 ),
490 ),473 ),
491 listThreadSessionRows(undefined).pipe(474 ),
492 Effect.mapError(475 listThreadActivityRows(undefined).pipe(
493 toPersistenceSqlOrDecodeError(476 Effect.mapError(
494 "ProjectionSnapshotQuery.getSnapshot:listThreadSessions:query",477 toPersistenceSqlOrDecodeError(
495 "ProjectionSnapshotQuery.getSnapshot:listThreadSessions:decodeRows",478 "ProjectionSnapshotQuery.getSnapshot:listThreadActivities:query",
496 ),479 "ProjectionSnapshotQuery.getSnapshot:listThreadActivities:decodeRows",
497 ),480 ),
498 ),481 ),
499 listCheckpointRows(undefined).pipe(482 ),
500 Effect.mapError(483 listThreadSessionRows(undefined).pipe(
501 toPersistenceSqlOrDecodeError(484 Effect.mapError(
502 "ProjectionSnapshotQuery.getSnapshot:listCheckpoints:query",485 toPersistenceSqlOrDecodeError(
503 "ProjectionSnapshotQuery.getSnapshot:listCheckpoints:decodeRows",486 "ProjectionSnapshotQuery.getSnapshot:listThreadSessions:query",
504 ),487 "ProjectionSnapshotQuery.getSnapshot:listThreadSessions:decodeRows",
505 ),488 ),
506 ),489 ),
507 listLatestTurnRows(undefined).pipe(490 ),
508 Effect.mapError(491 listCheckpointRows(undefined).pipe(
509 toPersistenceSqlOrDecodeError(492 Effect.mapError(
510 "ProjectionSnapshotQuery.getSnapshot:listLatestTurns:query",493 toPersistenceSqlOrDecodeError(
511 "ProjectionSnapshotQuery.getSnapshot:listLatestTurns:decodeRows",494 "ProjectionSnapshotQuery.getSnapshot:listCheckpoints:query",
512 ),495 "ProjectionSnapshotQuery.getSnapshot:listCheckpoints:decodeRows",
513 ),496 ),
514 ),497 ),
515 listProjectionStateRows(undefined).pipe(498 ),
516 Effect.mapError(499 listLatestTurnRows(undefined).pipe(
517 toPersistenceSqlOrDecodeError(500 Effect.mapError(
518 "ProjectionSnapshotQuery.getSnapshot:listProjectionState:query",501 toPersistenceSqlOrDecodeError(
519 "ProjectionSnapshotQuery.getSnapshot:listProjectionState:decodeRows",502 "ProjectionSnapshotQuery.getSnapshot:listLatestTurns:query",
520 ),503 "ProjectionSnapshotQuery.getSnapshot:listLatestTurns:decodeRows",
521 ),504 ),
522 ),505 ),
523 ]);506 ),
524 507 listProjectionStateRows(undefined).pipe(
525 const messagesByThread = new Map<string, Array<OrchestrationMessage>>();
526 const proposedPlansByThread = new Map<string, Array<OrchestrationProposedPlan>>();
527 const activitiesByThread = new Map<string, Array<OrchestrationThreadActivity>>();
528 const checkpointsByThread = new Map<string, Array<OrchestrationCheckpointSummary>>();
529 const sessionsByThread = new Map<string, OrchestrationSession>();
530 const latestTurnByThread = new Map<string, OrchestrationLatestTurn>();
531
532 let updatedAt: string | null = null;
533
534 for (const row of projectRows) {
535 updatedAt = maxIso(updatedAt, row.updatedAt);
536 }
537 for (const row of threadRows) {
538 updatedAt = maxIso(updatedAt, row.updatedAt);
539 }
540 for (const row of stateRows) {
541 updatedAt = maxIso(updatedAt, row.updatedAt);
542 }
543
544 for (const row of messageRows) {
545 updatedAt = maxIso(updatedAt, row.updatedAt);
546 const threadMessages = messagesByThread.get(row.threadId) ?? [];
547 threadMessages.push({
548 id: row.messageId,
549 role: row.role,
550 text: row.text,
551 ...(row.attachments !== null ? { attachments: row.attachments } : {}),
552 turnId: row.turnId,
553 streaming: row.isStreaming === 1,
554 createdAt: row.createdAt,
555 updatedAt: row.updatedAt,
556 });
557 messagesByThread.set(row.threadId, threadMessages);
558 }
559
560 for (const row of proposedPlanRows) {
561 updatedAt = maxIso(updatedAt, row.updatedAt);
562 const threadProposedPlans = proposedPlansByThread.get(row.threadId) ?? [];
563 threadProposedPlans.push({
564 id: row.planId,
565 turnId: row.turnId,
566 planMarkdown: row.planMarkdown,
567 implementedAt: row.implementedAt,
568 implementationThreadId: row.implementationThreadId,
569 createdAt: row.createdAt,
570 updatedAt: row.updatedAt,
571 });
572 proposedPlansByThread.set(row.threadId, threadProposedPlans);
573 }
574
575 for (const row of activityRows) {
576 updatedAt = maxIso(updatedAt, row.createdAt);
577 const threadActivities = activitiesByThread.get(row.threadId) ?? [];
578 threadActivities.push({
579 id: row.activityId,
580 tone: row.tone,
581 kind: row.kind,
582 summary: row.summary,
583 payload: row.payload,
584 turnId: row.turnId,
585 ...(row.sequence !== null ? { sequence: row.sequence } : {}),
586 createdAt: row.createdAt,
587 });
588 activitiesByThread.set(row.threadId, threadActivities);
589 }
590
591 for (const row of checkpointRows) {
592 updatedAt = maxIso(updatedAt, row.completedAt);
593 const threadCheckpoints = checkpointsByThread.get(row.threadId) ?? [];
594 threadCheckpoints.push({
595 turnId: row.turnId,
596 checkpointTurnCount: row.checkpointTurnCount,
597 checkpointRef: row.checkpointRef,
598 status: row.status,
599 files: row.files,
600 assistantMessageId: row.assistantMessageId,
601 completedAt: row.completedAt,
602 });
603 checkpointsByThread.set(row.threadId, threadCheckpoints);
604 }
605
606 for (const row of latestTurnRows) {
607 updatedAt = maxIso(updatedAt, row.requestedAt);
608 if (row.startedAt !== null) {
609 updatedAt = maxIso(updatedAt, row.startedAt);
610 }
611 if (row.completedAt !== null) {
612 updatedAt = maxIso(updatedAt, row.completedAt);
613 }
614 if (latestTurnByThread.has(row.threadId)) {
615 continue;
616 }
617 latestTurnByThread.set(row.threadId, {
618 turnId: row.turnId,
619 state:
620 row.state === "error"
621 ? "error"
622 : row.state === "interrupted"
623 ? "interrupted"
624 : row.state === "completed"
625 ? "completed"
626 : "running",
627 requestedAt: row.requestedAt,
628 startedAt: row.startedAt,
629 completedAt: row.completedAt,
630 assistantMessageId: row.assistantMessageId,
631 ...(row.sourceProposedPlanThreadId !== null && row.sourceProposedPlanId !== null
632 ? {
633 sourceProposedPlan: {
634 threadId: row.sourceProposedPlanThreadId,
635 planId: row.sourceProposedPlanId,
636 },
637 }
638 : {}),
639 });
640 }
641
642 for (const row of sessionRows) {
643 updatedAt = maxIso(updatedAt, row.updatedAt);
644 sessionsByThread.set(row.threadId, {
645 threadId: row.threadId,
646 status: row.status,
647 providerName: row.providerName,
648 runtimeMode: row.runtimeMode,
649 activeTurnId: row.activeTurnId,
650 lastError: row.lastError,
651 updatedAt: row.updatedAt,
652 });
653 }
654
655 const projects: ReadonlyArray<OrchestrationProject> = projectRows.map((row) => ({
656 id: row.projectId,
657 title: row.title,
658 workspaceRoot: row.workspaceRoot,
659 defaultModelSelection: row.defaultModelSelection,
660 scripts: row.scripts,
661 createdAt: row.createdAt,
662 updatedAt: row.updatedAt,
663 deletedAt: row.deletedAt,
664 }));
665
666 const threads: ReadonlyArray<OrchestrationThread> = threadRows.map((row) => ({
667 id: row.threadId,
668 projectId: row.projectId,
669 title: row.title,
670 modelSelection: row.modelSelection,
671 runtimeMode: row.runtimeMode,
672 interactionMode: row.interactionMode,
673 branch: row.branch,
674 worktreePath: row.worktreePath,
675 latestTurn: latestTurnByThread.get(row.threadId) ?? null,
676 createdAt: row.createdAt,
677 updatedAt: row.updatedAt,
678 archivedAt: row.archivedAt,
679 deletedAt: row.deletedAt,
680 messages: messagesByThread.get(row.threadId) ?? [],
681 proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
682 activities: activitiesByThread.get(row.threadId) ?? [],
683 checkpoints: checkpointsByThread.get(row.threadId) ?? [],
684 session: sessionsByThread.get(row.threadId) ?? null,
685 }));
686
687 const snapshot = {
688 snapshotSequence: computeSnapshotSequence(stateRows),
689 projects,
690 threads,
691 updatedAt: updatedAt ?? new Date(0).toISOString(),
692 };
693
694 return yield* decodeReadModel(snapshot).pipe(
695 Effect.mapError(508 Effect.mapError(
696 toPersistenceDecodeError("ProjectionSnapshotQuery.getSnapshot:decodeReadModel"),509 toPersistenceSqlOrDecodeError(
510 "ProjectionSnapshotQuery.getSnapshot:listProjectionState:query",
511 "ProjectionSnapshotQuery.getSnapshot:listProjectionState:decodeRows",
512 ),
697 ),513 ),
698 );514 ),
699 }),515 ]),
700 )516 )
701 .pipe(517 .pipe(
518 Effect.flatMap(
519 ([
520 projectRows,
521 threadRows,
522 messageRows,
523 proposedPlanRows,
524 activityRows,
525 sessionRows,
526 checkpointRows,
527 latestTurnRows,
528 stateRows,
529 ]) =>
530 Effect.gen(function* () {
531 const messagesByThread = new Map<string, Array<OrchestrationMessage>>();
532 const proposedPlansByThread = new Map<string, Array<OrchestrationProposedPlan>>();
533 const activitiesByThread = new Map<string, Array<OrchestrationThreadActivity>>();
534 const checkpointsByThread = new Map<string, Array<OrchestrationCheckpointSummary>>();
535 const sessionsByThread = new Map<string, OrchestrationSession>();
536 const latestTurnByThread = new Map<string, OrchestrationLatestTurn>();
537
538 let updatedAt: string | null = null;
539
540 for (const row of projectRows) {
541 updatedAt = maxIso(updatedAt, row.updatedAt);
542 }
543 for (const row of threadRows) {
544 updatedAt = maxIso(updatedAt, row.updatedAt);
545 }
546 for (const row of stateRows) {
547 updatedAt = maxIso(updatedAt, row.updatedAt);
548 }
549
550 for (const row of messageRows) {
551 updatedAt = maxIso(updatedAt, row.updatedAt);
552 const threadMessages = messagesByThread.get(row.threadId) ?? [];
553 threadMessages.push({
554 id: row.messageId,
555 role: row.role,
556 text: row.text,
557 ...(row.attachments !== null ? { attachments: row.attachments } : {}),
558 turnId: row.turnId,
559 streaming: row.isStreaming === 1,
560 createdAt: row.createdAt,
561 updatedAt: row.updatedAt,
562 });
563 messagesByThread.set(row.threadId, threadMessages);
564 }
565
566 for (const row of proposedPlanRows) {
567 updatedAt = maxIso(updatedAt, row.updatedAt);
568 const threadProposedPlans = proposedPlansByThread.get(row.threadId) ?? [];
569 threadProposedPlans.push({
570 id: row.planId,
571 turnId: row.turnId,
572 planMarkdown: row.planMarkdown,
573 implementedAt: row.implementedAt,
574 implementationThreadId: row.implementationThreadId,
575 createdAt: row.createdAt,
576 updatedAt: row.updatedAt,
577 });
578 proposedPlansByThread.set(row.threadId, threadProposedPlans);
579 }
580
581 for (const row of activityRows) {
582 updatedAt = maxIso(updatedAt, row.createdAt);
583 const threadActivities = activitiesByThread.get(row.threadId) ?? [];
584 threadActivities.push({
585 id: row.activityId,
586 tone: row.tone,
587 kind: row.kind,
588 summary: row.summary,
589 payload: row.payload,
590 turnId: row.turnId,
591 ...(row.sequence !== null ? { sequence: row.sequence } : {}),
592 createdAt: row.createdAt,
593 });
594 activitiesByThread.set(row.threadId, threadActivities);
595 }
596
597 for (const row of checkpointRows) {
598 updatedAt = maxIso(updatedAt, row.completedAt);
599 const threadCheckpoints = checkpointsByThread.get(row.threadId) ?? [];
600 threadCheckpoints.push({
601 turnId: row.turnId,
602 checkpointTurnCount: row.checkpointTurnCount,
603 checkpointRef: row.checkpointRef,
604 status: row.status,
605 files: row.files,
606 assistantMessageId: row.assistantMessageId,
607 completedAt: row.completedAt,
608 });
609 checkpointsByThread.set(row.threadId, threadCheckpoints);
610 }
611
612 for (const row of latestTurnRows) {
613 updatedAt = maxIso(updatedAt, row.requestedAt);
614 if (row.startedAt !== null) {
615 updatedAt = maxIso(updatedAt, row.startedAt);
616 }
617 if (row.completedAt !== null) {
618 updatedAt = maxIso(updatedAt, row.completedAt);
619 }
620 if (latestTurnByThread.has(row.threadId)) {
621 continue;
622 }
623 latestTurnByThread.set(row.threadId, {
624 turnId: row.turnId,
625 state:
626 row.state === "error"
627 ? "error"
628 : row.state === "interrupted"
629 ? "interrupted"
630 : row.state === "completed"
631 ? "completed"
632 : "running",
633 requestedAt: row.requestedAt,
634 startedAt: row.startedAt,
635 completedAt: row.completedAt,
636 assistantMessageId: row.assistantMessageId,
637 ...(row.sourceProposedPlanThreadId !== null && row.sourceProposedPlanId !== null
638 ? {
639 sourceProposedPlan: {
640 threadId: row.sourceProposedPlanThreadId,
641 planId: row.sourceProposedPlanId,
642 },
643 }
644 : {}),
645 });
646 }
647
648 for (const row of sessionRows) {
649 updatedAt = maxIso(updatedAt, row.updatedAt);
650 sessionsByThread.set(row.threadId, {
651 threadId: row.threadId,
652 status: row.status,
653 providerName: row.providerName,
654 runtimeMode: row.runtimeMode,
655 activeTurnId: row.activeTurnId,
656 lastError: row.lastError,
657 updatedAt: row.updatedAt,
658 });
659 }
660
661 const repositoryIdentities = new Map(
662 yield* Effect.forEach(
663 projectRows,
664 (row) =>
665 repositoryIdentityResolver
666 .resolve(row.workspaceRoot)
667 .pipe(Effect.map((identity) => [row.projectId, identity] as const)),
668 { concurrency: repositoryIdentityResolutionConcurrency },
669 ),
670 );
671
672 const projects: ReadonlyArray<OrchestrationProject> = projectRows.map((row) => ({
673 id: row.projectId,
674 title: row.title,
675 workspaceRoot: row.workspaceRoot,
676 repositoryIdentity: repositoryIdentities.get(row.projectId) ?? null,
677 defaultModelSelection: row.defaultModelSelection,
678 scripts: row.scripts,
679 createdAt: row.createdAt,
680 updatedAt: row.updatedAt,
681 deletedAt: row.deletedAt,
682 }));
683
684 const threads: ReadonlyArray<OrchestrationThread> = threadRows.map((row) => ({
685 id: row.threadId,
686 projectId: row.projectId,
687 title: row.title,
688 modelSelection: row.modelSelection,
689 runtimeMode: row.runtimeMode,
690 interactionMode: row.interactionMode,
691 branch: row.branch,
692 worktreePath: row.worktreePath,
693 latestTurn: latestTurnByThread.get(row.threadId) ?? null,
694 createdAt: row.createdAt,
695 updatedAt: row.updatedAt,
696 archivedAt: row.archivedAt,
697 deletedAt: row.deletedAt,
698 messages: messagesByThread.get(row.threadId) ?? [],
699 proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
700 activities: activitiesByThread.get(row.threadId) ?? [],
701 checkpoints: checkpointsByThread.get(row.threadId) ?? [],
702 session: sessionsByThread.get(row.threadId) ?? null,
703 }));
704
705 const snapshot = {
706 snapshotSequence: computeSnapshotSequence(stateRows),
707 projects,
708 threads,
709 updatedAt: updatedAt ?? new Date(0).toISOString(),
710 };
711
712 return yield* decodeReadModel(snapshot).pipe(
713 Effect.mapError(
714 toPersistenceDecodeError("ProjectionSnapshotQuery.getSnapshot:decodeReadModel"),
715 ),
716 );
717 }),
718 ),
702 Effect.mapError((error) => {719 Effect.mapError((error) => {
703 if (isPersistenceError(error)) {720 if (isPersistenceError(error)) {
704 return error;721 return error;
@@ -732,19 +749,24 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { · 27 unmodified lines
732 "ProjectionSnapshotQuery.getActiveProjectByWorkspaceRoot:decodeRow",749 "ProjectionSnapshotQuery.getActiveProjectByWorkspaceRoot:decodeRow",
733 ),750 ),
734 ),751 ),
735 Effect.map(752 Effect.flatMap((option) =>
736 Option.map(753 Option.isNone(option)
737 (row): OrchestrationProject => ({754 ? Effect.succeed(Option.none<OrchestrationProject>())
738 id: row.projectId,755 : repositoryIdentityResolver.resolve(option.value.workspaceRoot).pipe(
739 title: row.title,756 Effect.map((repositoryIdentity) =>
740 workspaceRoot: row.workspaceRoot,757 Option.some({
741 defaultModelSelection: row.defaultModelSelection,758 id: option.value.projectId,
742 scripts: row.scripts,759 title: option.value.title,
743 createdAt: row.createdAt,760 workspaceRoot: option.value.workspaceRoot,
744 updatedAt: row.updatedAt,761 repositoryIdentity,
745 deletedAt: row.deletedAt,762 defaultModelSelection: option.value.defaultModelSelection,
746 }),763 scripts: option.value.scripts,
747 ),764 createdAt: option.value.createdAt,
765 updatedAt: option.value.updatedAt,
766 deletedAt: option.value.deletedAt,
767 } satisfies OrchestrationProject),
768 ),
769 ),
748 ),770 ),
749 );771 );
750 772
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+2−0
@@ -28,6 +28,7 @@ import { · 27 unmodified lines
28} from "../../provider/Services/ProviderService.ts";28} from "../../provider/Services/ProviderService.ts";
29import { GitCore, type GitCoreShape } from "../../git/Services/GitCore.ts";29import { GitCore, type GitCoreShape } from "../../git/Services/GitCore.ts";
30import { TextGeneration, type TextGenerationShape } from "../../git/Services/TextGeneration.ts";30import { TextGeneration, type TextGenerationShape } from "../../git/Services/TextGeneration.ts";
31import { RepositoryIdentityResolverLive } from "../../project/Layers/RepositoryIdentityResolver.ts";
31import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";32import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
32import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";33import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
33import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";34import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
@@ -218,6 +219,7 @@ describe("ProviderCommandReactor", () => { · 184 unmodified lines
218 Layer.provide(OrchestrationProjectionPipelineLive),219 Layer.provide(OrchestrationProjectionPipelineLive),
219 Layer.provide(OrchestrationEventStoreLive),220 Layer.provide(OrchestrationEventStoreLive),
220 Layer.provide(OrchestrationCommandReceiptRepositoryLive),221 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
222 Layer.provide(RepositoryIdentityResolverLive),
221 Layer.provide(SqlitePersistenceMemory),223 Layer.provide(SqlitePersistenceMemory),
222 );224 );
223 const layer = ProviderCommandReactorLive.pipe(225 const layer = ProviderCommandReactorLive.pipe(
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+2−0
@@ -29,6 +29,7 @@ import { · 28 unmodified lines
29 ProviderService,29 ProviderService,
30 type ProviderServiceShape,30 type ProviderServiceShape,
31} from "../../provider/Services/ProviderService.ts";31} from "../../provider/Services/ProviderService.ts";
32import { RepositoryIdentityResolverLive } from "../../project/Layers/RepositoryIdentityResolver.ts";
32import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";33import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
33import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";34import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts";
34import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";35import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
@@ -204,6 +205,7 @@ describe("ProviderRuntimeIngestion", () => { · 169 unmodified lines
204 Layer.provide(OrchestrationProjectionPipelineLive),205 Layer.provide(OrchestrationProjectionPipelineLive),
205 Layer.provide(OrchestrationEventStoreLive),206 Layer.provide(OrchestrationEventStoreLive),
206 Layer.provide(OrchestrationCommandReceiptRepositoryLive),207 Layer.provide(OrchestrationCommandReceiptRepositoryLive),
208 Layer.provide(RepositoryIdentityResolverLive),
207 Layer.provide(SqlitePersistenceMemory),209 Layer.provide(SqlitePersistenceMemory),
208 );210 );
209 const layer = ProviderRuntimeIngestionLive.pipe(211 const layer = ProviderRuntimeIngestionLive.pipe(
apps/server/src/project/Layers/RepositoryIdentityResolver.test.ts+193−0
@@ -0,0 +1,193 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { expect, it } from "@effect/vitest";
3import { Duration, Effect, FileSystem, Layer } from "effect";
4import { TestClock } from "effect/testing";
5
6import { runProcess } from "../../processRunner.ts";
7import { RepositoryIdentityResolver } from "../Services/RepositoryIdentityResolver.ts";
8import {
9 makeRepositoryIdentityResolver,
10 RepositoryIdentityResolverLive,
11} from "./RepositoryIdentityResolver.ts";
12
13const git = (cwd: string, args: ReadonlyArray<string>) =>
14 Effect.promise(() => runProcess("git", ["-C", cwd, ...args]));
15
16const makeRepositoryIdentityResolverTestLayer = (options: {
17 readonly positiveCacheTtl?: Duration.Input;
18 readonly negativeCacheTtl?: Duration.Input;
19}) =>
20 Layer.effect(
21 RepositoryIdentityResolver,
22 makeRepositoryIdentityResolver({
23 cacheCapacity: 16,
24 ...options,
25 }),
26 );
27
28it.layer(NodeServices.layer)("RepositoryIdentityResolverLive", (it) => {
29 it.effect("normalizes equivalent GitHub remotes into a stable repository identity", () =>
30 Effect.gen(function* () {
31 const fileSystem = yield* FileSystem.FileSystem;
32 const cwd = yield* fileSystem.makeTempDirectoryScoped({
33 prefix: "t3-repository-identity-test-",
34 });
35
36 yield* git(cwd, ["init"]);
37 yield* git(cwd, ["remote", "add", "origin", "git@github.com:T3Tools/t3code.git"]);
38
39 const resolver = yield* RepositoryIdentityResolver;
40 const identity = yield* resolver.resolve(cwd);
41
42 expect(identity).not.toBeNull();
43 expect(identity?.canonicalKey).toBe("github.com/t3tools/t3code");
44 expect(identity?.displayName).toBe("t3tools/t3code");
45 expect(identity?.provider).toBe("github");
46 expect(identity?.owner).toBe("t3tools");
47 expect(identity?.name).toBe("t3code");
48 }).pipe(Effect.provide(RepositoryIdentityResolverLive)),
49 );
50
51 it.effect("returns null for non-git folders and repos without remotes", () =>
52 Effect.gen(function* () {
53 const fileSystem = yield* FileSystem.FileSystem;
54 const nonGitDir = yield* fileSystem.makeTempDirectoryScoped({
55 prefix: "t3-repository-identity-non-git-",
56 });
57 const gitDir = yield* fileSystem.makeTempDirectoryScoped({
58 prefix: "t3-repository-identity-no-remote-",
59 });
60
61 yield* git(gitDir, ["init"]);
62
63 const resolver = yield* RepositoryIdentityResolver;
64 const nonGitIdentity = yield* resolver.resolve(nonGitDir);
65 const noRemoteIdentity = yield* resolver.resolve(gitDir);
66
67 expect(nonGitIdentity).toBeNull();
68 expect(noRemoteIdentity).toBeNull();
69 }).pipe(Effect.provide(RepositoryIdentityResolverLive)),
70 );
71
72 it.effect("prefers upstream over origin when both remotes are configured", () =>
73 Effect.gen(function* () {
74 const fileSystem = yield* FileSystem.FileSystem;
75 const cwd = yield* fileSystem.makeTempDirectoryScoped({
76 prefix: "t3-repository-identity-upstream-test-",
77 });
78
79 yield* git(cwd, ["init"]);
80 yield* git(cwd, ["remote", "add", "origin", "git@github.com:julius/t3code.git"]);
81 yield* git(cwd, ["remote", "add", "upstream", "git@github.com:T3Tools/t3code.git"]);
82
83 const resolver = yield* RepositoryIdentityResolver;
84 const identity = yield* resolver.resolve(cwd);
85
86 expect(identity).not.toBeNull();
87 expect(identity?.locator.remoteName).toBe("upstream");
88 expect(identity?.canonicalKey).toBe("github.com/t3tools/t3code");
89 expect(identity?.displayName).toBe("t3tools/t3code");
90 }).pipe(Effect.provide(RepositoryIdentityResolverLive)),
91 );
92
93 it.effect("uses the last remote path segment as the repository name for nested groups", () =>
94 Effect.gen(function* () {
95 const fileSystem = yield* FileSystem.FileSystem;
96 const cwd = yield* fileSystem.makeTempDirectoryScoped({
97 prefix: "t3-repository-identity-nested-group-test-",
98 });
99
100 yield* git(cwd, ["init"]);
101 yield* git(cwd, ["remote", "add", "origin", "git@gitlab.com:T3Tools/platform/t3code.git"]);
102
103 const resolver = yield* RepositoryIdentityResolver;
104 const identity = yield* resolver.resolve(cwd);
105
106 expect(identity).not.toBeNull();
107 expect(identity?.canonicalKey).toBe("gitlab.com/t3tools/platform/t3code");
108 expect(identity?.displayName).toBe("t3tools/platform/t3code");
109 expect(identity?.owner).toBe("t3tools");
110 expect(identity?.name).toBe("t3code");
111 }).pipe(Effect.provide(RepositoryIdentityResolverLive)),
112 );
113
114 it.effect(
115 "refreshes cached null identities after the negative TTL when a remote is configured later",
116 () =>
117 Effect.gen(function* () {
118 const fileSystem = yield* FileSystem.FileSystem;
119 const cwd = yield* fileSystem.makeTempDirectoryScoped({
120 prefix: "t3-repository-identity-late-remote-test-",
121 });
122
123 yield* git(cwd, ["init"]);
124
125 const resolver = yield* RepositoryIdentityResolver;
126 const initialIdentity = yield* resolver.resolve(cwd);
127 expect(initialIdentity).toBeNull();
128
129 yield* git(cwd, ["remote", "add", "origin", "git@github.com:T3Tools/t3code.git"]);
130
131 const cachedIdentity = yield* resolver.resolve(cwd);
132 expect(cachedIdentity).toBeNull();
133
134 yield* TestClock.adjust(Duration.millis(120));
135
136 const refreshedIdentity = yield* resolver.resolve(cwd);
137 expect(refreshedIdentity).not.toBeNull();
138 expect(refreshedIdentity?.canonicalKey).toBe("github.com/t3tools/t3code");
139 expect(refreshedIdentity?.name).toBe("t3code");
140 }).pipe(
141 Effect.provide(
142 Layer.merge(
143 TestClock.layer(),
144 makeRepositoryIdentityResolverTestLayer({
145 negativeCacheTtl: Duration.millis(50),
146 positiveCacheTtl: Duration.seconds(1),
147 }),
148 ),
149 ),
150 ),
151 );
152
153 it.effect("refreshes cached identities after the positive TTL when a remote changes", () =>
154 Effect.gen(function* () {
155 const fileSystem = yield* FileSystem.FileSystem;
156 const cwd = yield* fileSystem.makeTempDirectoryScoped({
157 prefix: "t3-repository-identity-remote-change-test-",
158 });
159
160 yield* git(cwd, ["init"]);
161 yield* git(cwd, ["remote", "add", "origin", "git@github.com:T3Tools/t3code.git"]);
162
163 const resolver = yield* RepositoryIdentityResolver;
164 const initialIdentity = yield* resolver.resolve(cwd);
165 expect(initialIdentity).not.toBeNull();
166 expect(initialIdentity?.canonicalKey).toBe("github.com/t3tools/t3code");
167
168 yield* git(cwd, ["remote", "set-url", "origin", "git@github.com:T3Tools/t3code-next.git"]);
169
170 const cachedIdentity = yield* resolver.resolve(cwd);
171 expect(cachedIdentity).not.toBeNull();
172 expect(cachedIdentity?.canonicalKey).toBe("github.com/t3tools/t3code");
173
174 yield* TestClock.adjust(Duration.millis(180));
175
176 const refreshedIdentity = yield* resolver.resolve(cwd);
177 expect(refreshedIdentity).not.toBeNull();
178 expect(refreshedIdentity?.canonicalKey).toBe("github.com/t3tools/t3code-next");
179 expect(refreshedIdentity?.displayName).toBe("t3tools/t3code-next");
180 expect(refreshedIdentity?.name).toBe("t3code-next");
181 }).pipe(
182 Effect.provide(
183 Layer.merge(
184 TestClock.layer(),
185 makeRepositoryIdentityResolverTestLayer({
186 negativeCacheTtl: Duration.millis(50),
187 positiveCacheTtl: Duration.millis(100),
188 }),
189 ),
190 ),
191 ),
192 );
193});
apps/server/src/project/Layers/RepositoryIdentityResolver.ts+147−0
@@ -0,0 +1,147 @@
1import type { RepositoryIdentity } from "@t3tools/contracts";
2import { Cache, Duration, Effect, Exit, Layer } from "effect";
3import { detectGitHostingProviderFromRemoteUrl, normalizeGitRemoteUrl } from "@t3tools/shared/git";
4
5import { runProcess } from "../../processRunner.ts";
6import {
7 RepositoryIdentityResolver,
8 type RepositoryIdentityResolverShape,
9} from "../Services/RepositoryIdentityResolver.ts";
10
11function parseRemoteFetchUrls(stdout: string): Map<string, string> {
12 const remotes = new Map<string, string>();
13 for (const line of stdout.split("\n")) {
14 const trimmed = line.trim();
15 if (trimmed.length === 0) continue;
16 const match = /^(\S+)\s+(\S+)\s+\((fetch|push)\)$/.exec(trimmed);
17 if (!match) continue;
18 const [, remoteName = "", remoteUrl = "", direction = ""] = match;
19 if (direction !== "fetch" || remoteName.length === 0 || remoteUrl.length === 0) {
20 continue;
21 }
22 remotes.set(remoteName, remoteUrl);
23 }
24 return remotes;
25}
26
27function pickPrimaryRemote(
28 remotes: ReadonlyMap<string, string>,
29): { readonly remoteName: string; readonly remoteUrl: string } | null {
30 for (const preferredRemoteName of ["upstream", "origin"] as const) {
31 const remoteUrl = remotes.get(preferredRemoteName);
32 if (remoteUrl) {
33 return { remoteName: preferredRemoteName, remoteUrl };
34 }
35 }
36
37 const [remoteName, remoteUrl] =
38 [...remotes.entries()].toSorted(([left], [right]) => left.localeCompare(right))[0] ?? [];
39 return remoteName && remoteUrl ? { remoteName, remoteUrl } : null;
40}
41
42function buildRepositoryIdentity(input: {
43 readonly remoteName: string;
44 readonly remoteUrl: string;
45}): RepositoryIdentity {
46 const canonicalKey = normalizeGitRemoteUrl(input.remoteUrl);
47 const hostingProvider = detectGitHostingProviderFromRemoteUrl(input.remoteUrl);
48 const repositoryPath = canonicalKey.split("/").slice(1).join("/");
49 const repositoryPathSegments = repositoryPath.split("/").filter((segment) => segment.length > 0);
50 const [owner] = repositoryPathSegments;
51 const repositoryName = repositoryPathSegments.at(-1);
52
53 return {
54 canonicalKey,
55 locator: {
56 source: "git-remote",
57 remoteName: input.remoteName,
58 remoteUrl: input.remoteUrl,
59 },
60 ...(repositoryPath ? { displayName: repositoryPath } : {}),
61 ...(hostingProvider ? { provider: hostingProvider.kind } : {}),
62 ...(owner ? { owner } : {}),
63 ...(repositoryName ? { name: repositoryName } : {}),
64 };
65}
66
67const DEFAULT_REPOSITORY_IDENTITY_CACHE_CAPACITY = 512;
68const DEFAULT_POSITIVE_CACHE_TTL = Duration.minutes(1);
69const DEFAULT_NEGATIVE_CACHE_TTL = Duration.seconds(10);
70
71interface RepositoryIdentityResolverOptions {
72 readonly cacheCapacity?: number;
73 readonly positiveCacheTtl?: Duration.Input;
74 readonly negativeCacheTtl?: Duration.Input;
75}
76
77async function resolveRepositoryIdentityCacheKey(cwd: string): Promise<string> {
78 let cacheKey = cwd;
79
80 try {
81 const topLevelResult = await runProcess("git", ["-C", cwd, "rev-parse", "--show-toplevel"], {
82 allowNonZeroExit: true,
83 });
84 if (topLevelResult.code !== 0) {
85 return cacheKey;
86 }
87
88 const candidate = topLevelResult.stdout.trim();
89 if (candidate.length > 0) {
90 cacheKey = candidate;
91 }
92 } catch {
93 return cacheKey;
94 }
95
96 return cacheKey;
97}
98
99async function resolveRepositoryIdentityFromCacheKey(
100 cacheKey: string,
101): Promise<RepositoryIdentity | null> {
102 try {
103 const remoteResult = await runProcess("git", ["-C", cacheKey, "remote", "-v"], {
104 allowNonZeroExit: true,
105 });
106 if (remoteResult.code !== 0) {
107 return null;
108 }
109
110 const remote = pickPrimaryRemote(parseRemoteFetchUrls(remoteResult.stdout));
111 return remote ? buildRepositoryIdentity(remote) : null;
112 } catch {
113 return null;
114 }
115}
116
117export const makeRepositoryIdentityResolver = Effect.fn("makeRepositoryIdentityResolver")(
118 function* (options: RepositoryIdentityResolverOptions = {}) {
119 const repositoryIdentityCache = yield* Cache.makeWith<string, RepositoryIdentity | null>({
120 capacity: options.cacheCapacity ?? DEFAULT_REPOSITORY_IDENTITY_CACHE_CAPACITY,
121 lookup: (cacheKey) => Effect.promise(() => resolveRepositoryIdentityFromCacheKey(cacheKey)),
122 timeToLive: Exit.match({
123 onSuccess: (value) =>
124 value === null
125 ? (options.negativeCacheTtl ?? DEFAULT_NEGATIVE_CACHE_TTL)
126 : (options.positiveCacheTtl ?? DEFAULT_POSITIVE_CACHE_TTL),
127 onFailure: () => Duration.zero,
128 }),
129 });
130
131 const resolve: RepositoryIdentityResolverShape["resolve"] = Effect.fn(
132 "RepositoryIdentityResolver.resolve",
133 )(function* (cwd) {
134 const cacheKey = yield* Effect.promise(() => resolveRepositoryIdentityCacheKey(cwd));
135 return yield* Cache.get(repositoryIdentityCache, cacheKey);
136 });
137
138 return {
139 resolve,
140 } satisfies RepositoryIdentityResolverShape;
141 },
142);
143
144export const RepositoryIdentityResolverLive = Layer.effect(
145 RepositoryIdentityResolver,
146 makeRepositoryIdentityResolver(),
147);
apps/server/src/project/Services/RepositoryIdentityResolver.ts+12−0
@@ -0,0 +1,12 @@
1import type { RepositoryIdentity } from "@t3tools/contracts";
2import { ServiceMap } from "effect";
3import type { Effect } from "effect";
4
5export interface RepositoryIdentityResolverShape {
6 readonly resolve: (cwd: string) => Effect.Effect<RepositoryIdentity | null>;
7}
8
9export class RepositoryIdentityResolver extends ServiceMap.Service<
10 RepositoryIdentityResolver,
11 RepositoryIdentityResolverShape
12>()("t3/project/Services/RepositoryIdentityResolver") {}
apps/server/src/server.test.ts+245−1
@@ -5,6 +5,8 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; · 4 unmodified lines
5import {5import {
6 CommandId,6 CommandId,
7 DEFAULT_SERVER_SETTINGS,7 DEFAULT_SERVER_SETTINGS,
8 EnvironmentId,
9 EventId,
8 GitCommandError,10 GitCommandError,
9 KeybindingRule,11 KeybindingRule,
10 MessageId,12 MessageId,
@@ -83,6 +85,14 @@ import { · 72 unmodified lines
83 ProjectSetupScriptRunner,85 ProjectSetupScriptRunner,
84 type ProjectSetupScriptRunnerShape,86 type ProjectSetupScriptRunnerShape,
85} from "./project/Services/ProjectSetupScriptRunner.ts";87} from "./project/Services/ProjectSetupScriptRunner.ts";
88import {
89 RepositoryIdentityResolver,
90 type RepositoryIdentityResolverShape,
91} from "./project/Services/RepositoryIdentityResolver.ts";
92import {
93 ServerEnvironment,
94 type ServerEnvironmentShape,
95} from "./environment/Services/ServerEnvironment.ts";
86import { WorkspaceEntriesLive } from "./workspace/Layers/WorkspaceEntries.ts";96import { WorkspaceEntriesLive } from "./workspace/Layers/WorkspaceEntries.ts";
87import { WorkspaceFileSystemLive } from "./workspace/Layers/WorkspaceFileSystem.ts";97import { WorkspaceFileSystemLive } from "./workspace/Layers/WorkspaceFileSystem.ts";
88import { WorkspacePathsLive } from "./workspace/Layers/WorkspacePaths.ts";98import { WorkspacePathsLive } from "./workspace/Layers/WorkspacePaths.ts";
@@ -93,6 +103,18 @@ const defaultModelSelection = { · 4 unmodified lines
93 provider: "codex",103 provider: "codex",
94 model: "gpt-5-codex",104 model: "gpt-5-codex",
95} as const;105} as const;
106const testEnvironmentDescriptor = {
107 environmentId: EnvironmentId.makeUnsafe("environment-test"),
108 label: "Test environment",
109 platform: {
110 os: "darwin" as const,
111 arch: "arm64" as const,
112 },
113 serverVersion: "0.0.0-test",
114 capabilities: {
115 repositoryIdentity: true,
116 },
117};
96 118
97const makeDefaultOrchestrationReadModel = () => {119const makeDefaultOrchestrationReadModel = () => {
98 const now = new Date().toISOString();120 const now = new Date().toISOString();
@@ -270,6 +292,8 @@ const buildAppUnderTest = (options?: { · 171 unmodified lines
270 browserTraceCollector?: Partial<BrowserTraceCollectorShape>;292 browserTraceCollector?: Partial<BrowserTraceCollectorShape>;
271 serverLifecycleEvents?: Partial<ServerLifecycleEventsShape>;293 serverLifecycleEvents?: Partial<ServerLifecycleEventsShape>;
272 serverRuntimeStartup?: Partial<ServerRuntimeStartupShape>;294 serverRuntimeStartup?: Partial<ServerRuntimeStartupShape>;
295 serverEnvironment?: Partial<ServerEnvironmentShape>;
296 repositoryIdentityResolver?: Partial<RepositoryIdentityResolverShape>;
273 };297 };
274}) =>298}) =>
275 Effect.gen(function* () {299 Effect.gen(function* () {
@@ -416,6 +440,19 @@ const buildAppUnderTest = (options?: { · 140 unmodified lines
416 ...options?.layers?.serverRuntimeStartup,440 ...options?.layers?.serverRuntimeStartup,
417 }),441 }),
418 ),442 ),
443 Layer.provide(
444 Layer.mock(ServerEnvironment)({
445 getEnvironmentId: Effect.succeed(testEnvironmentDescriptor.environmentId),
446 getDescriptor: Effect.succeed(testEnvironmentDescriptor),
447 ...options?.layers?.serverEnvironment,
448 }),
449 ),
450 Layer.provide(
451 Layer.mock(RepositoryIdentityResolver)({
452 resolve: () => Effect.succeed(null),
453 ...options?.layers?.repositoryIdentityResolver,
454 }),
455 ),
419 Layer.provide(workspaceAndProjectServicesLayer),456 Layer.provide(workspaceAndProjectServicesLayer),
420 Layer.provideMerge(FetchHttpClient.layer),457 Layer.provideMerge(FetchHttpClient.layer),
421 Layer.provide(layerConfig),458 Layer.provide(layerConfig),
@@ -1069,6 +1106,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { · 647 unmodified lines
1069 sequence: 1,1106 sequence: 1,
1070 type: "welcome" as const,1107 type: "welcome" as const,
1071 payload: {1108 payload: {
1109 environment: testEnvironmentDescriptor,
1072 cwd: "/tmp/project",1110 cwd: "/tmp/project",
1073 projectName: "project",1111 projectName: "project",
1074 },1112 },
@@ -1078,7 +1116,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { · 3 unmodified lines
1078 version: 1 as const,1116 version: 1 as const,
1079 sequence: 2,1117 sequence: 2,
1080 type: "ready" as const,1118 type: "ready" as const,
1081 payload: { at: new Date().toISOString() },1119 payload: { at: new Date().toISOString(), environment: testEnvironmentDescriptor },
1082 });1120 });
1083 1121
1084 yield* buildAppUnderTest({1122 yield* buildAppUnderTest({
@@ -1996,6 +2034,73 @@ it.layer(NodeServices.layer)("server router seam", (it) => { · 911 unmodified lines
1996 }).pipe(Effect.provide(NodeHttpServer.layerTest)),2034 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
1997 );2035 );
1998 2036
2037 it.effect("enriches replayed project events with repository identity metadata", () =>
2038 Effect.gen(function* () {
2039 const repositoryIdentity = {
2040 canonicalKey: "github.com/t3tools/t3code",
2041 locator: {
2042 source: "git-remote" as const,
2043 remoteName: "origin",
2044 remoteUrl: "git@github.com:T3Tools/t3code.git",
2045 },
2046 displayName: "T3Tools/t3code",
2047 provider: "github",
2048 owner: "T3Tools",
2049 name: "t3code",
2050 };
2051
2052 yield* buildAppUnderTest({
2053 layers: {
2054 orchestrationEngine: {
2055 readEvents: (_fromSequenceExclusive) =>
2056 Stream.make({
2057 sequence: 1,
2058 eventId: EventId.makeUnsafe("event-1"),
2059 aggregateKind: "project",
2060 aggregateId: defaultProjectId,
2061 occurredAt: "2026-04-05T00:00:00.000Z",
2062 commandId: null,
2063 causationEventId: null,
2064 correlationId: null,
2065 metadata: {},
2066 type: "project.created",
2067 payload: {
2068 projectId: defaultProjectId,
2069 title: "Default Project",
2070 workspaceRoot: "/tmp/default-project",
2071 defaultModelSelection,
2072 scripts: [],
2073 createdAt: "2026-04-05T00:00:00.000Z",
2074 updatedAt: "2026-04-05T00:00:00.000Z",
2075 },
2076 } satisfies Extract<OrchestrationEvent, { type: "project.created" }>),
2077 },
2078 repositoryIdentityResolver: {
2079 resolve: () => Effect.succeed(repositoryIdentity),
2080 },
2081 },
2082 });
2083
2084 const wsUrl = yield* getWsServerUrl("/ws");
2085 const replayResult = yield* Effect.scoped(
2086 withWsRpcClient(wsUrl, (client) =>
2087 client[ORCHESTRATION_WS_METHODS.replayEvents]({
2088 fromSequenceExclusive: 0,
2089 }),
2090 ),
2091 );
2092
2093 const replayedEvent = replayResult[0];
2094 assert.equal(replayedEvent?.type, "project.created");
2095 assert.deepEqual(
2096 replayedEvent && replayedEvent.type === "project.created"
2097 ? replayedEvent.payload.repositoryIdentity
2098 : null,
2099 repositoryIdentity,
2100 );
2101 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
2102 );
2103
1999 it.effect("closes thread terminals after a successful archive command", () =>2104 it.effect("closes thread terminals after a successful archive command", () =>
2000 Effect.gen(function* () {2105 Effect.gen(function* () {
2001 const threadId = ThreadId.makeUnsafe("thread-archive");2106 const threadId = ThreadId.makeUnsafe("thread-archive");
@@ -2498,6 +2603,145 @@ it.layer(NodeServices.layer)("server router seam", (it) => { · 496 unmodified lines
2498 }).pipe(Effect.provide(NodeHttpServer.layerTest)),2603 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
2499 );2604 );
2500 2605
2606 it.effect("enriches replayed project events only once before streaming them to subscribers", () =>
2607 Effect.gen(function* () {
2608 let resolveCalls = 0;
2609 const repositoryIdentity = {
2610 canonicalKey: "github.com/t3tools/t3code",
2611 locator: {
2612 source: "git-remote" as const,
2613 remoteName: "origin",
2614 remoteUrl: "git@github.com:t3tools/t3code.git",
2615 },
2616 displayName: "t3tools/t3code",
2617 provider: "github" as const,
2618 owner: "t3tools",
2619 name: "t3code",
2620 };
2621
2622 yield* buildAppUnderTest({
2623 layers: {
2624 orchestrationEngine: {
2625 getReadModel: () =>
2626 Effect.succeed({
2627 ...makeDefaultOrchestrationReadModel(),
2628 snapshotSequence: 0,
2629 }),
2630 readEvents: () =>
2631 Stream.make({
2632 sequence: 1,
2633 eventId: EventId.makeUnsafe("event-1"),
2634 aggregateKind: "project",
2635 aggregateId: defaultProjectId,
2636 occurredAt: "2026-04-06T00:00:00.000Z",
2637 commandId: null,
2638 causationEventId: null,
2639 correlationId: null,
2640 metadata: {},
2641 type: "project.meta-updated",
2642 payload: {
2643 projectId: defaultProjectId,
2644 title: "Replayed Project",
2645 updatedAt: "2026-04-06T00:00:00.000Z",
2646 },
2647 } satisfies Extract<OrchestrationEvent, { type: "project.meta-updated" }>),
2648 streamDomainEvents: Stream.empty,
2649 },
2650 repositoryIdentityResolver: {
2651 resolve: () => {
2652 resolveCalls += 1;
2653 return Effect.succeed(repositoryIdentity);
2654 },
2655 },
2656 },
2657 });
2658
2659 const wsUrl = yield* getWsServerUrl("/ws");
2660 const events = yield* Effect.scoped(
2661 withWsRpcClient(wsUrl, (client) =>
2662 client[WS_METHODS.subscribeOrchestrationDomainEvents]({}).pipe(
2663 Stream.take(1),
2664 Stream.runCollect,
2665 ),
2666 ),
2667 );
2668
2669 const event = Array.from(events)[0];
2670 assert.equal(resolveCalls, 1);
2671 assert.equal(event?.type, "project.meta-updated");
2672 assert.deepEqual(
2673 event && event.type === "project.meta-updated" ? event.payload.repositoryIdentity : null,
2674 repositoryIdentity,
2675 );
2676 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
2677 );
2678
2679 it.effect("enriches subscribed project meta updates with repository identity metadata", () =>
2680 Effect.gen(function* () {
2681 const repositoryIdentity = {
2682 canonicalKey: "github.com/t3tools/t3code",
2683 locator: {
2684 source: "git-remote" as const,
2685 remoteName: "upstream",
2686 remoteUrl: "git@github.com:T3Tools/t3code.git",
2687 },
2688 displayName: "T3Tools/t3code",
2689 provider: "github",
2690 owner: "T3Tools",
2691 name: "t3code",
2692 };
2693
2694 yield* buildAppUnderTest({
2695 layers: {
2696 orchestrationEngine: {
2697 getReadModel: () =>
2698 Effect.succeed({
2699 ...makeDefaultOrchestrationReadModel(),
2700 snapshotSequence: 0,
2701 }),
2702 streamDomainEvents: Stream.make({
2703 sequence: 1,
2704 eventId: EventId.makeUnsafe("event-1"),
2705 aggregateKind: "project",
2706 aggregateId: defaultProjectId,
2707 occurredAt: "2026-04-05T00:00:00.000Z",
2708 commandId: null,
2709 causationEventId: null,
2710 correlationId: null,
2711 metadata: {},
2712 type: "project.meta-updated",
2713 payload: {
2714 projectId: defaultProjectId,
2715 title: "Renamed Project",
2716 updatedAt: "2026-04-05T00:00:00.000Z",
2717 },
2718 } satisfies Extract<OrchestrationEvent, { type: "project.meta-updated" }>),
2719 },
2720 repositoryIdentityResolver: {
2721 resolve: () => Effect.succeed(repositoryIdentity),
2722 },
2723 },
2724 });
2725
2726 const wsUrl = yield* getWsServerUrl("/ws");
2727 const events = yield* Effect.scoped(
2728 withWsRpcClient(wsUrl, (client) =>
2729 client[WS_METHODS.subscribeOrchestrationDomainEvents]({}).pipe(
2730 Stream.take(1),
2731 Stream.runCollect,
2732 ),
2733 ),
2734 );
2735
2736 const event = Array.from(events)[0];
2737 assert.equal(event?.type, "project.meta-updated");
2738 assert.deepEqual(
2739 event && event.type === "project.meta-updated" ? event.payload.repositoryIdentity : null,
2740 repositoryIdentity,
2741 );
2742 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
2743 );
2744
2501 it.effect("routes websocket rpc orchestration.getSnapshot errors", () =>2745 it.effect("routes websocket rpc orchestration.getSnapshot errors", () =>
2502 Effect.gen(function* () {2746 Effect.gen(function* () {
2503 yield* buildAppUnderTest({2747 yield* buildAppUnderTest({
apps/server/src/server.ts+4−0
@@ -44,11 +44,13 @@ import { CheckpointReactorLive } from "./orchestration/Layers/CheckpointReactor" · 43 unmodified lines
44import { ProviderRegistryLive } from "./provider/Layers/ProviderRegistry";44import { ProviderRegistryLive } from "./provider/Layers/ProviderRegistry";
45import { ServerSettingsLive } from "./serverSettings";45import { ServerSettingsLive } from "./serverSettings";
46import { ProjectFaviconResolverLive } from "./project/Layers/ProjectFaviconResolver";46import { ProjectFaviconResolverLive } from "./project/Layers/ProjectFaviconResolver";
47import { RepositoryIdentityResolverLive } from "./project/Layers/RepositoryIdentityResolver";
47import { WorkspaceEntriesLive } from "./workspace/Layers/WorkspaceEntries";48import { WorkspaceEntriesLive } from "./workspace/Layers/WorkspaceEntries";
48import { WorkspaceFileSystemLive } from "./workspace/Layers/WorkspaceFileSystem";49import { WorkspaceFileSystemLive } from "./workspace/Layers/WorkspaceFileSystem";
49import { WorkspacePathsLive } from "./workspace/Layers/WorkspacePaths";50import { WorkspacePathsLive } from "./workspace/Layers/WorkspacePaths";
50import { ProjectSetupScriptRunnerLive } from "./project/Layers/ProjectSetupScriptRunner";51import { ProjectSetupScriptRunnerLive } from "./project/Layers/ProjectSetupScriptRunner";
51import { ObservabilityLive } from "./observability/Layers/Observability";52import { ObservabilityLive } from "./observability/Layers/Observability";
53import { ServerEnvironmentLive } from "./environment/Layers/ServerEnvironment";
52 54
53const PtyAdapterLive = Layer.unwrap(55const PtyAdapterLive = Layer.unwrap(
54 Effect.gen(function* () {56 Effect.gen(function* () {
@@ -199,6 +201,8 @@ const RuntimeDependenciesLive = ReactorLayerLive.pipe( · 144 unmodified lines
199 Layer.provideMerge(ServerSettingsLive),201 Layer.provideMerge(ServerSettingsLive),
200 Layer.provideMerge(WorkspaceLayerLive),202 Layer.provideMerge(WorkspaceLayerLive),
201 Layer.provideMerge(ProjectFaviconResolverLive),203 Layer.provideMerge(ProjectFaviconResolverLive),
204 Layer.provideMerge(RepositoryIdentityResolverLive),
205 Layer.provideMerge(ServerEnvironmentLive),
202 206
203 // Misc.207 // Misc.
204 Layer.provideMerge(AnalyticsServiceLayerLive),208 Layer.provideMerge(AnalyticsServiceLayerLive),
apps/server/src/serverLifecycleEvents.test.ts+10−0
@@ -1,3 +1,4 @@
1import { EnvironmentId } from "@t3tools/contracts";
1import { assert, it } from "@effect/vitest";2import { assert, it } from "@effect/vitest";
2import { assertTrue } from "@effect/vitest/utils";3import { assertTrue } from "@effect/vitest/utils";
3import { Effect, Option } from "effect";4import { Effect, Option } from "effect";
@@ -9,12 +10,20 @@ it.effect( · 5 unmodified lines
9 () =>10 () =>
10 Effect.gen(function* () {11 Effect.gen(function* () {
11 const lifecycleEvents = yield* ServerLifecycleEvents;12 const lifecycleEvents = yield* ServerLifecycleEvents;
13 const environment = {
14 environmentId: EnvironmentId.makeUnsafe("environment-test"),
15 label: "Test environment",
16 platform: { os: "darwin" as const, arch: "arm64" as const },
17 serverVersion: "0.0.0-test",
18 capabilities: { repositoryIdentity: true },
19 };
12 20
13 const welcome = yield* lifecycleEvents21 const welcome = yield* lifecycleEvents
14 .publish({22 .publish({
15 version: 1,23 version: 1,
16 type: "welcome",24 type: "welcome",
17 payload: {25 payload: {
26 environment,
18 cwd: "/tmp/project",27 cwd: "/tmp/project",
19 projectName: "project",28 projectName: "project",
20 },29 },
@@ -29,6 +38,7 @@ it.effect( · 8 unmodified lines
29 type: "ready",38 type: "ready",
30 payload: {39 payload: {
31 at: new Date().toISOString(),40 at: new Date().toISOString(),
41 environment,
32 },42 },
33 })43 })
34 .pipe(Effect.timeoutOption("50 millis"));44 .pipe(Effect.timeoutOption("50 millis"));
apps/server/src/serverRuntimeStartup.ts+12−2
@@ -27,6 +27,7 @@ import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnap · 26 unmodified lines
27import { OrchestrationReactor } from "./orchestration/Services/OrchestrationReactor";27import { OrchestrationReactor } from "./orchestration/Services/OrchestrationReactor";
28import { ServerLifecycleEvents } from "./serverLifecycleEvents";28import { ServerLifecycleEvents } from "./serverLifecycleEvents";
29import { ServerSettingsService } from "./serverSettings";29import { ServerSettingsService } from "./serverSettings";
30import { ServerEnvironment } from "./environment/Services/ServerEnvironment";
30import { AnalyticsService } from "./telemetry/Services/AnalyticsService";31import { AnalyticsService } from "./telemetry/Services/AnalyticsService";
31 32
32const isWildcardHost = (host: string | undefined): boolean =>33const isWildcardHost = (host: string | undefined): boolean =>
@@ -262,6 +263,7 @@ const makeServerRuntimeStartup = Effect.gen(function* () { · 229 unmodified lines
262 const orchestrationReactor = yield* OrchestrationReactor;263 const orchestrationReactor = yield* OrchestrationReactor;
263 const lifecycleEvents = yield* ServerLifecycleEvents;264 const lifecycleEvents = yield* ServerLifecycleEvents;
264 const serverSettings = yield* ServerSettingsService;265 const serverSettings = yield* ServerSettingsService;
266 const serverEnvironment = yield* ServerEnvironment;
265 267
266 const commandGate = yield* makeCommandGate;268 const commandGate = yield* makeCommandGate;
267 const httpListening = yield* Deferred.make<void>();269 const httpListening = yield* Deferred.make<void>();
@@ -308,7 +310,9 @@ const makeServerRuntimeStartup = Effect.gen(function* () { · 40 unmodified lines
308 310
309 yield* Effect.logDebug("startup phase: preparing welcome payload");311 yield* Effect.logDebug("startup phase: preparing welcome payload");
310 const welcome = yield* runStartupPhase("welcome.prepare", autoBootstrapWelcome);312 const welcome = yield* runStartupPhase("welcome.prepare", autoBootstrapWelcome);
313 const environment = yield* serverEnvironment.getDescriptor;
311 yield* Effect.logDebug("startup phase: publishing welcome event", {314 yield* Effect.logDebug("startup phase: publishing welcome event", {
315 environmentId: environment.environmentId,
312 cwd: welcome.cwd,316 cwd: welcome.cwd,
313 projectName: welcome.projectName,317 projectName: welcome.projectName,
314 bootstrapProjectId: welcome.bootstrapProjectId,318 bootstrapProjectId: welcome.bootstrapProjectId,
@@ -319,7 +323,10 @@ const makeServerRuntimeStartup = Effect.gen(function* () { · 4 unmodified lines
319 lifecycleEvents.publish({323 lifecycleEvents.publish({
320 version: 1,324 version: 1,
321 type: "welcome",325 type: "welcome",
322 payload: welcome,326 payload: {
327 environment,
328 ...welcome,
329 },
323 }),330 }),
324 );331 );
325 }).pipe(332 }).pipe(
@@ -354,7 +361,10 @@ const makeServerRuntimeStartup = Effect.gen(function* () { · 28 unmodified lines
354 lifecycleEvents.publish({361 lifecycleEvents.publish({
355 version: 1,362 version: 1,
356 type: "ready",363 type: "ready",
357 payload: { at: new Date().toISOString() },364 payload: {
365 at: new Date().toISOString(),
366 environment: yield* serverEnvironment.getDescriptor,
367 },
358 }),368 }),
359 );369 );
360 370
apps/server/src/ws.ts+55−1
@@ -47,6 +47,8 @@ import { WorkspaceEntries } from "./workspace/Services/WorkspaceEntries"; · 46 unmodified lines
47import { WorkspaceFileSystem } from "./workspace/Services/WorkspaceFileSystem";47import { WorkspaceFileSystem } from "./workspace/Services/WorkspaceFileSystem";
48import { WorkspacePathOutsideRootError } from "./workspace/Services/WorkspacePaths";48import { WorkspacePathOutsideRootError } from "./workspace/Services/WorkspacePaths";
49import { ProjectSetupScriptRunner } from "./project/Services/ProjectSetupScriptRunner";49import { ProjectSetupScriptRunner } from "./project/Services/ProjectSetupScriptRunner";
50import { RepositoryIdentityResolver } from "./project/Services/RepositoryIdentityResolver";
51import { ServerEnvironment } from "./environment/Services/ServerEnvironment";
50 52
51const WsRpcLayer = WsRpcGroup.toLayer(53const WsRpcLayer = WsRpcGroup.toLayer(
52 Effect.gen(function* () {54 Effect.gen(function* () {
@@ -67,6 +69,8 @@ const WsRpcLayer = WsRpcGroup.toLayer( · 14 unmodified lines
67 const workspaceEntries = yield* WorkspaceEntries;69 const workspaceEntries = yield* WorkspaceEntries;
68 const workspaceFileSystem = yield* WorkspaceFileSystem;70 const workspaceFileSystem = yield* WorkspaceFileSystem;
69 const projectSetupScriptRunner = yield* ProjectSetupScriptRunner;71 const projectSetupScriptRunner = yield* ProjectSetupScriptRunner;
72 const repositoryIdentityResolver = yield* RepositoryIdentityResolver;
73 const serverEnvironment = yield* ServerEnvironment;
70 const serverCommandId = (tag: string) =>74 const serverCommandId = (tag: string) =>
71 CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`);75 CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`);
72 76
@@ -113,6 +117,49 @@ const WsRpcLayer = WsRpcGroup.toLayer( · 40 unmodified lines
113 });117 });
114 };118 };
115 119
120 const enrichProjectEvent = (
121 event: OrchestrationEvent,
122 ): Effect.Effect<OrchestrationEvent, never, never> => {
123 switch (event.type) {
124 case "project.created":
125 return repositoryIdentityResolver.resolve(event.payload.workspaceRoot).pipe(
126 Effect.map((repositoryIdentity) => ({
127 ...event,
128 payload: {
129 ...event.payload,
130 repositoryIdentity,
131 },
132 })),
133 );
134 case "project.meta-updated":
135 return Effect.gen(function* () {
136 const workspaceRoot =
137 event.payload.workspaceRoot ??
138 (yield* orchestrationEngine.getReadModel()).projects.find(
139 (project) => project.id === event.payload.projectId,
140 )?.workspaceRoot ??
141 null;
142 if (workspaceRoot === null) {
143 return event;
144 }
145
146 const repositoryIdentity = yield* repositoryIdentityResolver.resolve(workspaceRoot);
147 return {
148 ...event,
149 payload: {
150 ...event.payload,
151 repositoryIdentity,
152 },
153 } satisfies OrchestrationEvent;
154 });
155 default:
156 return Effect.succeed(event);
157 }
158 };
159
160 const enrichOrchestrationEvents = (events: ReadonlyArray<OrchestrationEvent>) =>
161 Effect.forEach(events, enrichProjectEvent, { concurrency: 4 });
162
116 const dispatchBootstrapTurnStart = (163 const dispatchBootstrapTurnStart = (
117 command: Extract<OrchestrationCommand, { type: "thread.turn.start" }>,164 command: Extract<OrchestrationCommand, { type: "thread.turn.start" }>,
118 ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> =>165 ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> =>
@@ -329,8 +376,10 @@ const WsRpcLayer = WsRpcGroup.toLayer( · 210 unmodified lines
329 const keybindingsConfig = yield* keybindings.loadConfigState;376 const keybindingsConfig = yield* keybindings.loadConfigState;
330 const providers = yield* providerRegistry.getProviders;377 const providers = yield* providerRegistry.getProviders;
331 const settings = yield* serverSettings.getSettings;378 const settings = yield* serverSettings.getSettings;
379 const environment = yield* serverEnvironment.getDescriptor;
332 380
333 return {381 return {
382 environment,
334 cwd: config.cwd,383 cwd: config.cwd,
335 keybindingsConfigPath: config.keybindingsConfigPath,384 keybindingsConfigPath: config.keybindingsConfigPath,
336 keybindings: keybindingsConfig.keybindings,385 keybindings: keybindingsConfig.keybindings,
@@ -435,6 +484,7 @@ const WsRpcLayer = WsRpcGroup.toLayer( · 98 unmodified lines
435 ),484 ),
436 ).pipe(485 ).pipe(
437 Effect.map((events) => Array.from(events)),486 Effect.map((events) => Array.from(events)),
487 Effect.flatMap(enrichOrchestrationEvents),
438 Effect.mapError(488 Effect.mapError(
439 (cause) =>489 (cause) =>
440 new OrchestrationReplayEventsError({490 new OrchestrationReplayEventsError({
@@ -455,10 +505,14 @@ const WsRpcLayer = WsRpcGroup.toLayer( · 14 unmodified lines
455 orchestrationEngine.readEvents(fromSequenceExclusive),505 orchestrationEngine.readEvents(fromSequenceExclusive),
456 ).pipe(506 ).pipe(
457 Effect.map((events) => Array.from(events)),507 Effect.map((events) => Array.from(events)),
508 Effect.flatMap(enrichOrchestrationEvents),
458 Effect.catch(() => Effect.succeed([] as Array<OrchestrationEvent>)),509 Effect.catch(() => Effect.succeed([] as Array<OrchestrationEvent>)),
459 );510 );
460 const replayStream = Stream.fromIterable(replayEvents);511 const replayStream = Stream.fromIterable(replayEvents);
461 const source = Stream.merge(replayStream, orchestrationEngine.streamDomainEvents);512 const liveStream = orchestrationEngine.streamDomainEvents.pipe(
513 Stream.mapEffect(enrichProjectEvent),
514 );
515 const source = Stream.merge(replayStream, liveStream);
462 type SequenceState = {516 type SequenceState = {
463 readonly nextSequence: number;517 readonly nextSequence: number;
464 readonly pendingBySequence: Map<number, OrchestrationEvent>;518 readonly pendingBySequence: Map<number, OrchestrationEvent>;