import { AuthOrchestrationOperateScope, AuthOrchestrationReadScope, EnvironmentHttpApi,} from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as Option from "effect/Option";import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder"; import { projectThreadDetailSnapshot } from "./ActivityPayloadProjection.ts";import { cleanupFailedUploadedAttachments, normalizeDispatchCommand } from "./Normalizer.ts";import { annotateEnvironmentRequest, failEnvironmentInternal, failEnvironmentInvalidRequest, failEnvironmentNotFound, requireEnvironmentScope,} from "../auth/http.ts";import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts";import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts"; export const orchestrationHttpApiLayer = HttpApiBuilder.group( EnvironmentHttpApi, "orchestration", Effect.fnUntraced(function* (handlers) { const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; const orchestrationEngine = yield* OrchestrationEngineService; return handlers .handle( "snapshot", Effect.fn("environment.orchestration.snapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); // Serve the lightweight command read model (thread bodies empty) // instead of the fully hydrated snapshot. Hydrating every message // and activity payload in the database has OOM-killed servers, and // the route's only consumer (the project CLI) reads projects alone — // UI clients load the shell and per-thread snapshots instead. return yield* projectionSnapshotQuery .getCommandReadModel() .pipe( Effect.catch((cause) => failEnvironmentInternal("orchestration_snapshot_failed", cause), ), ); }), ) .handle( "shellSnapshot", Effect.fn("environment.orchestration.shellSnapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); return yield* projectionSnapshotQuery .getShellSnapshot() .pipe( Effect.catch((cause) => failEnvironmentInternal("orchestration_snapshot_failed", cause), ), ); }), ) .handle( "threadSnapshot", Effect.fn("environment.orchestration.threadSnapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); const snapshot = yield* projectionSnapshotQuery .getThreadDetailSnapshot( args.params.threadId, args.payload.turnLimit === undefined ? undefined : { turnLimit: args.payload.turnLimit, ...(args.payload.beforeCursor !== undefined ? { beforeCursor: args.payload.beforeCursor } : {}), }, ) .pipe( Effect.catch((cause) => failEnvironmentInternal("orchestration_thread_snapshot_failed", cause), ), ); if (Option.isNone(snapshot)) { return yield* failEnvironmentNotFound("thread_not_found"); } return projectThreadDetailSnapshot(snapshot.value); }), ) .handle( "dispatch", Effect.fn("environment.orchestration.dispatch")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationOperateScope); const normalizedCommand = yield* normalizeDispatchCommand(args.payload).pipe( Effect.catch(() => failEnvironmentInvalidRequest("invalid_command")), ); return yield* orchestrationEngine.dispatch(normalizedCommand).pipe( Effect.tapError(() => cleanupFailedUploadedAttachments(args.payload, normalizedCommand), ), Effect.catch((cause) => failEnvironmentInternal("orchestration_dispatch_failed", cause), ), ); }), ); }),);
import {
AuthOrchestrationOperateScope,
AuthOrchestrationReadScope,
EnvironmentHttpApi,
} from "@t3tools/contracts";
import * as Effect from "effect/Effect";
import * as Option from "effect/Option";
import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder";
import { projectThreadDetailSnapshot } from "./ActivityPayloadProjection.ts";
import { cleanupFailedUploadedAttachments, normalizeDispatchCommand } from "./Normalizer.ts";
import {
annotateEnvironmentRequest,
failEnvironmentInternal,
failEnvironmentInvalidRequest,
failEnvironmentNotFound,
requireEnvironmentScope,
} from "../auth/http.ts";
import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts";
import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts";
export const orchestrationHttpApiLayer = HttpApiBuilder.group(
EnvironmentHttpApi,
"orchestration",
Effect.fnUntraced(function* (handlers) {
const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;
const orchestrationEngine = yield* OrchestrationEngineService;
return handlers
.handle(
"snapshot",
Effect.fn("environment.orchestration.snapshot")(function* (args) {
yield* annotateEnvironmentRequest(args.endpoint.name);
yield* requireEnvironmentScope(AuthOrchestrationReadScope);
// Serve the lightweight command read model (thread bodies empty)
// instead of the fully hydrated snapshot. Hydrating every message
// and activity payload in the database has OOM-killed servers, and
// the route's only consumer (the project CLI) reads projects alone —
// UI clients load the shell and per-thread snapshots instead.
return yield* projectionSnapshotQuery
.getCommandReadModel()
.pipe(
Effect.catch((cause) =>
failEnvironmentInternal("orchestration_snapshot_failed", cause),
),
);
}),
)
.handle(
"shellSnapshot",
Effect.fn("environment.orchestration.shellSnapshot")(function* (args) {
yield* annotateEnvironmentRequest(args.endpoint.name);
yield* requireEnvironmentScope(AuthOrchestrationReadScope);
return yield* projectionSnapshotQuery
.getShellSnapshot()
.pipe(
Effect.catch((cause) =>
failEnvironmentInternal("orchestration_snapshot_failed", cause),
),
);
}),
)
.handle(
"threadSnapshot",
Effect.fn("environment.orchestration.threadSnapshot")(function* (args) {
yield* annotateEnvironmentRequest(args.endpoint.name);
yield* requireEnvironmentScope(AuthOrchestrationReadScope);
const snapshot = yield* projectionSnapshotQuery
.getThreadDetailSnapshot(
args.params.threadId,
args.payload.turnLimit === undefined
? undefined
: {
turnLimit: args.payload.turnLimit,
...(args.payload.beforeCursor !== undefined
? { beforeCursor: args.payload.beforeCursor }
: {}),
},
)
.pipe(
Effect.catch((cause) =>
failEnvironmentInternal("orchestration_thread_snapshot_failed", cause),
),
);
if (Option.isNone(snapshot)) {
return yield* failEnvironmentNotFound("thread_not_found");
}
return projectThreadDetailSnapshot(snapshot.value);
}),
)
.handle(
"dispatch",
Effect.fn("environment.orchestration.dispatch")(function* (args) {
yield* annotateEnvironmentRequest(args.endpoint.name);
yield* requireEnvironmentScope(AuthOrchestrationOperateScope);
const normalizedCommand = yield* normalizeDispatchCommand(args.payload).pipe(
Effect.catch(() => failEnvironmentInvalidRequest("invalid_command")),
);
return yield* orchestrationEngine.dispatch(normalizedCommand).pipe(
Effect.tapError(() =>
cleanupFailedUploadedAttachments(args.payload, normalizedCommand),
),
Effect.catch((cause) =>
failEnvironmentInternal("orchestration_dispatch_failed", cause),
),
);
}),
);
}),
);