Skip to content
This repository was archived by the owner on Oct 10, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 61 additions & 14 deletions apps/server/integration/OrchestrationEngineHarness.integration.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import {
ApprovalRequestId,
CodexSettings,
ProviderDriverKind,
ThreadId,
type OrchestrationEvent,
type OrchestrationThread,
} from "@t3tools/contracts";
Expand All @@ -16,6 +17,7 @@ import * as Layer from "effect/Layer";
import * as ManagedRuntime from "effect/ManagedRuntime";
import * as Option from "effect/Option";
import * as Path from "effect/Path";
import * as PubSub from "effect/PubSub";
import * as Ref from "effect/Ref";
import * as Schedule from "effect/Schedule";
import * as Schema from "effect/Schema";
Expand Down Expand Up @@ -46,11 +48,10 @@ import {
import { ProviderService } from "../src/provider/Services/ProviderService.ts";
import { AnalyticsService } from "../src/telemetry/Services/AnalyticsService.ts";
import { CheckpointReactorLive } from "../src/orchestration/Layers/CheckpointReactor.ts";
import { RepositoryIdentityResolverLive } from "../src/project/Layers/RepositoryIdentityResolver.ts";
import { RepositoryIdentityResolver } from "../src/project/Services/RepositoryIdentityResolver.ts";
import { OrchestrationEngineLive } from "../src/orchestration/Layers/OrchestrationEngine.ts";
import { OrchestrationProjectionPipelineLive } from "../src/orchestration/Layers/ProjectionPipeline.ts";
import { OrchestrationProjectionSnapshotQueryLive } from "../src/orchestration/Layers/ProjectionSnapshotQuery.ts";
import { RuntimeReceiptBusTest } from "../src/orchestration/Layers/RuntimeReceiptBus.ts";
import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts";
import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts";
import { ProviderRuntimeIngestionLive } from "../src/orchestration/Layers/ProviderRuntimeIngestion.ts";
Expand Down Expand Up @@ -244,6 +245,13 @@ export const makeOrchestrationIntegrationHarness = (
makeAdapterRegistryMock({ [adapterHarness.provider]: adapterHarness.adapter }),
)
: null;
const receiptPubSub = yield* PubSub.unbounded<OrchestrationRuntimeReceipt>();
const runtimeReceiptBusLayer = Layer.succeed(RuntimeReceiptBus, {
publish: (receipt) => PubSub.publish(receiptPubSub, receipt).pipe(Effect.asVoid),
get streamEventsForTest() {
return Stream.fromPubSub(receiptPubSub);
},
});
const rootDir = yield* fileSystem.makeTempDirectoryScoped({
prefix: "t3-orchestration-integration-",
});
Expand Down Expand Up @@ -302,7 +310,7 @@ export const makeOrchestrationIntegrationHarness = (
ProjectionPendingApprovalRepositoryLive,
checkpointStoreLayer,
providerLayer,
RuntimeReceiptBusTest,
runtimeReceiptBusLayer,
);
const serverSettingsLayer = ServerSettingsService.layerTest();
const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe(
Expand Down Expand Up @@ -369,7 +377,11 @@ export const makeOrchestrationIntegrationHarness = (
Layer.provideMerge(runtimeServicesLayer),
Layer.provideMerge(orchestrationReactorLayer),
Layer.provide(persistenceLayer),
Layer.provideMerge(RepositoryIdentityResolverLive),
Layer.provideMerge(
Layer.succeed(RepositoryIdentityResolver, {
resolve: () => Effect.succeed(null),
}),
),
Layer.provideMerge(ServerSettingsService.layerTest()),
Layer.provideMerge(ServerConfig.layerTest(workspaceDir, rootDir)),
Layer.provideMerge(NodeServices.layer),
Expand Down Expand Up @@ -420,12 +432,8 @@ export const makeOrchestrationIntegrationHarness = (
) =>
waitFor(
snapshotQuery
.getSnapshot()
.pipe(
Effect.map(
(snapshot) => snapshot.threads.find((thread) => thread.id === threadId) ?? null,
),
),
.getThreadDetailById(ThreadId.make(threadId))
.pipe(Effect.map((thread) => Option.getOrNull(thread))),
(thread): thread is OrchestrationThread => thread !== null && predicate(thread),
`projected thread '${threadId}'`,
timeoutMs,
Expand Down Expand Up @@ -494,9 +502,40 @@ export const makeOrchestrationIntegrationHarness = (
predicate: (receipt: OrchestrationRuntimeReceipt) => boolean,
timeoutMs?: number,
) {
const readMatchingReceipt = Ref.get(receiptHistory).pipe(
Effect.map((history) => history.find(predicate)),
);
const readMatchingReceipt = Effect.gen(function* () {
const inMemoryReceipt = (yield* Ref.get(receiptHistory)).find(predicate);
if (inMemoryReceipt) {
return inMemoryReceipt;
}

const events = Array.from(yield* Stream.runCollect(engine.readEvents(0)));
const durableReceipts = events.flatMap(
(event): ReadonlyArray<OrchestrationRuntimeReceipt> => {
if (event.type !== "thread.turn-diff-completed") {
return [];
}
return [
{
type: "checkpoint.diff.finalized",
threadId: event.payload.threadId,
turnId: event.payload.turnId,
checkpointTurnCount: event.payload.checkpointTurnCount,
checkpointRef: event.payload.checkpointRef,
status: event.payload.status,
createdAt: event.payload.completedAt,
},
{
type: "turn.processing.quiesced",
threadId: event.payload.threadId,
turnId: event.payload.turnId,
checkpointTurnCount: event.payload.checkpointTurnCount,
createdAt: event.payload.completedAt,
},
];
},
);
return durableReceipts.find(predicate);
Comment thread
coderabbitai[bot] marked this conversation as resolved.
});

return waitFor(
readMatchingReceipt,
Expand Down Expand Up @@ -528,7 +567,15 @@ export const makeOrchestrationIntegrationHarness = (
}
});

yield* shutdown;
yield* shutdown.pipe(
Effect.timeout("5 seconds"),
Effect.catchCause((cause) =>
Effect.logWarning("orchestration integration harness disposal timed out", {
cause,
}),
),
Effect.asVoid,
);
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
});

return {
Expand Down
1 change: 1 addition & 0 deletions apps/server/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
"@effect/platform-node": "catalog:",
"@effect/platform-node-shared": "catalog:",
"@effect/sql-sqlite-bun": "catalog:",
"@factory/droid-sdk": "^0.2.0",
"@github/copilot": "1.0.2",
"@github/copilot-sdk": "^0.1.32",
"@opencode-ai/sdk": "^1.3.15",
Expand Down
157 changes: 157 additions & 0 deletions apps/server/src/provider/Drivers/DroidDriver.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,157 @@
import {
DroidSettings,
ProviderDriverKind,
TextGenerationError,
type ServerProvider,
} from "@t3tools/contracts";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as FileSystem from "effect/FileSystem";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";
import { HttpClient } from "effect/unstable/http";
import { ChildProcessSpawner } from "effect/unstable/process";

import { ServerConfig } from "../../config.ts";
import type { TextGenerationShape } from "../../textGeneration/TextGeneration.ts";
import { ProviderDriverError } from "../Errors.ts";
import { makeDroidAdapter } from "../Layers/DroidAdapter.ts";
import { checkDroidProviderStatus, makePendingDroidProvider } from "../Layers/DroidProvider.ts";
import { makeManagedServerProvider } from "../makeManagedServerProvider.ts";
import {
defaultProviderContinuationIdentity,
type ProviderDriver,
type ProviderInstance,
} from "../ProviderDriver.ts";
import type { ServerProviderDraft } from "../providerSnapshot.ts";
import { mergeProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts";
import {
enrichProviderSnapshotWithVersionAdvisory,
makePackageManagedProviderMaintenanceResolver,
resolveProviderMaintenanceCapabilitiesEffect,
} from "../providerMaintenance.ts";

const decodeDroidSettings = Schema.decodeSync(DroidSettings);
const DRIVER_KIND = ProviderDriverKind.make("droid");
const SNAPSHOT_REFRESH_INTERVAL = Duration.minutes(5);
const UPDATE = makePackageManagedProviderMaintenanceResolver({
provider: DRIVER_KIND,
npmPackageName: "droid",
homebrewFormula: null,
nativeUpdate: null,
});

export type DroidDriverEnv =
| ChildProcessSpawner.ChildProcessSpawner
| FileSystem.FileSystem
| HttpClient.HttpClient
| ServerConfig;

const withInstanceIdentity =
(input: {
readonly instanceId: ProviderInstance["instanceId"];
readonly displayName: string | undefined;
readonly accentColor: string | undefined;
readonly continuationGroupKey: string;
}) =>
(snapshot: ServerProviderDraft): ServerProvider => ({
...snapshot,
instanceId: input.instanceId,
driver: DRIVER_KIND,
...(input.displayName ? { displayName: input.displayName } : {}),
...(input.accentColor ? { accentColor: input.accentColor } : {}),
continuation: { groupKey: input.continuationGroupKey },
});

function makeUnsupportedTextGeneration(): TextGenerationShape {
const fail = (operation: TextGenerationError["operation"]) =>
Effect.fail(
new TextGenerationError({
operation,
detail: "Droid SDK text generation is not enabled in this WIP.",
}),
);
return {
generateCommitMessage: () => fail("generateCommitMessage"),
generatePrContent: () => fail("generatePrContent"),
generateBranchName: () => fail("generateBranchName"),
generateThreadTitle: () => fail("generateThreadTitle"),
};
}

export const DroidDriver: ProviderDriver<DroidSettings, DroidDriverEnv> = {
driverKind: DRIVER_KIND,
metadata: {
displayName: "Droid",
supportsMultipleInstances: true,
},
configSchema: DroidSettings,
defaultConfig: (): DroidSettings => decodeDroidSettings({}),
create: ({ instanceId, displayName, accentColor, environment, enabled, config }) =>
Effect.gen(function* () {
const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
const httpClient = yield* HttpClient.HttpClient;
const processEnv = mergeProviderInstanceEnvironment(environment);
const continuationIdentity = defaultProviderContinuationIdentity({
driverKind: DRIVER_KIND,
instanceId,
});
const stampIdentity = withInstanceIdentity({
instanceId,
displayName,
accentColor,
continuationGroupKey: continuationIdentity.continuationKey,
});
const effectiveConfig = { ...config, enabled } satisfies DroidSettings;
const maintenanceCapabilities = yield* resolveProviderMaintenanceCapabilitiesEffect(UPDATE, {
binaryPath: effectiveConfig.binaryPath,
env: processEnv,
});

const adapter = yield* makeDroidAdapter(effectiveConfig, {
instanceId,
environment: processEnv,
});
const checkProvider = checkDroidProviderStatus(effectiveConfig, processEnv).pipe(
Effect.map(stampIdentity),
Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
);
const snapshot = yield* makeManagedServerProvider<DroidSettings>({
maintenanceCapabilities,
getSettings: Effect.succeed(effectiveConfig),
streamSettings: Stream.never,
haveSettingsChanged: () => false,
initialSnapshot: (settings) =>
makePendingDroidProvider(settings).pipe(Effect.map(stampIdentity)),
checkProvider,
enrichSnapshot: ({ snapshot, publishSnapshot }) =>
enrichProviderSnapshotWithVersionAdvisory(snapshot, maintenanceCapabilities).pipe(
Effect.provideService(HttpClient.HttpClient, httpClient),
Effect.flatMap((enrichedSnapshot) => publishSnapshot(enrichedSnapshot)),
),
refreshInterval: SNAPSHOT_REFRESH_INTERVAL,
}).pipe(
Effect.mapError(
(cause) =>
new ProviderDriverError({
driver: DRIVER_KIND,
instanceId,
detail: `Failed to build Droid snapshot: ${cause.message ?? String(cause)}`,
cause,
}),
),
);

return {
instanceId,
driverKind: DRIVER_KIND,
continuationIdentity,
displayName,
accentColor,
enabled,
snapshot,
adapter,
textGeneration: makeUnsupportedTextGeneration(),
} satisfies ProviderInstance;
}),
};
2 changes: 2 additions & 0 deletions apps/server/src/provider/Layers/CodexSessionRuntime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -269,6 +269,7 @@ function runtimeModeToThreadConfig(input: RuntimeMode): {
sandbox: "read-only",
};
case "auto-accept-edits":
case "medium-access":
return {
approvalPolicy: "on-request",
sandbox: "workspace-write",
Expand Down Expand Up @@ -307,6 +308,7 @@ function runtimeModeToTurnSandboxPolicy(
type: "readOnly",
};
case "auto-accept-edits":
case "medium-access":
return {
type: "workspaceWrite",
};
Expand Down
Loading
Loading