Skip to content
Closed
Show file tree
Hide file tree
Changes from all 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
5 changes: 5 additions & 0 deletions apps/server/src/checkpointing/CheckpointDiffQuery.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@ describe("CheckpointDiffQuery.layer", () => {
getThreadRuntimeContext: () => Effect.die("unused"),
getTurnStartMessage: () => Effect.die("unused"),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadGroupShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
searchThreads: () => Effect.succeed({ matches: [] }),
Expand Down Expand Up @@ -214,6 +215,7 @@ describe("CheckpointDiffQuery.layer", () => {
getThreadRuntimeContext: () => Effect.die("unused"),
getTurnStartMessage: () => Effect.die("unused"),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadGroupShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
searchThreads: () => Effect.succeed({ matches: [] }),
Expand Down Expand Up @@ -305,6 +307,7 @@ describe("CheckpointDiffQuery.layer", () => {
getThreadRuntimeContext: () => Effect.die("unused"),
getTurnStartMessage: () => Effect.die("unused"),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadGroupShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
searchThreads: () => Effect.succeed({ matches: [] }),
Expand Down Expand Up @@ -381,6 +384,7 @@ describe("CheckpointDiffQuery.layer", () => {
getThreadRuntimeContext: () => Effect.die("unused"),
getTurnStartMessage: () => Effect.die("unused"),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadGroupShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
searchThreads: () => Effect.succeed({ matches: [] }),
Expand Down Expand Up @@ -442,6 +446,7 @@ describe("CheckpointDiffQuery.layer", () => {
getThreadRuntimeContext: () => Effect.die("unused"),
getTurnStartMessage: () => Effect.die("unused"),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadGroupShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
searchThreads: () => Effect.succeed({ matches: [] }),
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/environment/ServerEnvironment.ts
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,7 @@ export const make = Effect.gen(function* () {
threadPinReorder: true,
threadActiveReorder: true,
threadTitleRegeneration: true,
threadGroups: true,
threadPullRequests: true,
pullRequestStackActions: true,
threadPullRequestLinking: true,
Expand Down
15 changes: 15 additions & 0 deletions apps/server/src/git/GitManager.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -309,6 +309,10 @@ function createTextGeneration(
Effect.succeed({
title: "Update workflow",
}),
generateThreadGroupName: () =>
Effect.succeed({
name: "Workflow",
}),
...overrides,
};

Expand Down Expand Up @@ -357,6 +361,17 @@ function createTextGeneration(
}),
),
),
generateThreadGroupName: (input) =>
implementation.generateThreadGroupName(input).pipe(
Effect.mapError(
(cause) =>
new TextGenerationError({
operation: "generateThreadGroupName",
detail: "fake text generation failed",
...(cause !== undefined ? { cause } : {}),
}),
),
),
};
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -455,6 +455,7 @@ describe("OrchestrationEngine", () => {
getThreadRuntimeContext: () => Effect.die("unused"),
getTurnStartMessage: () => Effect.die("unused"),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadGroupShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
searchThreads: () => Effect.succeed({ matches: [] }),
Expand Down
19 changes: 15 additions & 4 deletions apps/server/src/orchestration/Layers/OrchestrationEngine.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import type {
OrchestrationEvent,
OrchestrationReadModel,
ProjectId,
ThreadGroupId,
ThreadId,
} from "@t3tools/contracts";
import { OrchestrationCommand } from "@t3tools/contracts";
Expand Down Expand Up @@ -62,8 +63,8 @@ interface CommandEnvelope {
}

function commandToAggregateRef(command: OrchestrationCommand): {
readonly aggregateKind: "project" | "thread";
readonly aggregateId: ProjectId | ThreadId;
readonly aggregateKind: "project" | "thread" | "thread-group";
readonly aggregateId: ProjectId | ThreadId | ThreadGroupId;
} {
switch (command.type) {
case "project.create":
Expand All @@ -73,6 +74,14 @@ function commandToAggregateRef(command: OrchestrationCommand): {
aggregateKind: "project",
aggregateId: command.projectId,
};
case "thread-group.create":
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
case "thread-group.meta.update":
case "thread-group.delete":
case "thread-group.name.generate.complete":
return {
aggregateKind: "thread-group",
aggregateId: command.groupId,
};
default:
return {
aggregateKind: "thread",
Expand Down Expand Up @@ -293,10 +302,12 @@ const makeOrchestrationEngine = Effect.gen(function* () {
});
}

// Keyed by the command's own aggregate: group commands also emit
// thread events, and a retry is matched against the command.
yield* commandReceiptRepository.upsert({
commandId: envelope.command.commandId,
aggregateKind: lastSavedEvent.aggregateKind,
aggregateId: lastSavedEvent.aggregateId,
aggregateKind: aggregateRef.aggregateKind,
aggregateId: aggregateRef.aggregateId,
acceptedAt: lastSavedEvent.occurredAt,
resultSequence: lastSavedEvent.sequence,
status: "accepted",
Expand Down
72 changes: 72 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionPipeline.ts
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ import {
type ProjectionThreadProposedPlan,
ProjectionThreadProposedPlanRepository,
} from "../../persistence/Services/ProjectionThreadProposedPlans.ts";
import * as ProjectionThreadGroups from "../../persistence/ProjectionThreadGroups.ts";
import * as ProjectionThreadPullRequests from "../../persistence/ProjectionThreadPullRequests.ts";
import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts";
import {
Expand Down Expand Up @@ -482,6 +483,8 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
const eventStore = yield* OrchestrationEventStore;
const projectionStateRepository = yield* ProjectionStateRepository;
const projectionProjectRepository = yield* ProjectionProjectRepository;
const projectionThreadGroupRepository =
yield* ProjectionThreadGroups.ProjectionThreadGroupRepository;
const projectionThreadRepository = yield* ProjectionThreadRepository;
const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository;
const projectionThreadProposedPlanRepository = yield* ProjectionThreadProposedPlanRepository;
Expand Down Expand Up @@ -564,6 +567,59 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
return;
}

// Thread groups ride the projects projector: they are project-scoped
// rows with the same lifecycle shape and no per-thread fan-out.
case "thread-group.created":
yield* projectionThreadGroupRepository.upsert({
groupId: event.payload.groupId,
projectId: event.payload.projectId,
name: event.payload.name,
icon: event.payload.icon,
nameGenerationRequestId: event.payload.nameGeneration?.requestId ?? null,
nameGenerationStartedAt: event.payload.nameGeneration?.startedAt ?? null,
createdAt: event.payload.createdAt,
updatedAt: event.payload.updatedAt,
deletedAt: null,
});
return;

case "thread-group.meta-updated": {
const existingRow = yield* projectionThreadGroupRepository.getById({
groupId: event.payload.groupId,
});
if (Option.isNone(existingRow)) {
return;
}
yield* projectionThreadGroupRepository.upsert({
...existingRow.value,
...(event.payload.name !== undefined ? { name: event.payload.name } : {}),
...(event.payload.icon !== undefined ? { icon: event.payload.icon } : {}),
...(event.payload.nameGeneration !== undefined
? {
nameGenerationRequestId: event.payload.nameGeneration?.requestId ?? null,
nameGenerationStartedAt: event.payload.nameGeneration?.startedAt ?? null,
}
: {}),
updatedAt: event.payload.updatedAt,
});
return;
}

case "thread-group.deleted": {
const existingRow = yield* projectionThreadGroupRepository.getById({
groupId: event.payload.groupId,
});
if (Option.isNone(existingRow)) {
return;
}
yield* projectionThreadGroupRepository.upsert({
...existingRow.value,
deletedAt: event.payload.deletedAt,
updatedAt: event.payload.deletedAt,
});
return;
}

default:
return;
}
Expand Down Expand Up @@ -797,6 +853,21 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
return;
}

case "thread.group-set": {
const existingRow = yield* projectionThreadRepository.getById({
threadId: event.payload.threadId,
});
if (Option.isNone(existingRow)) {
return;
}
yield* projectionThreadRepository.upsert({
...existingRow.value,
groupId: event.payload.groupId,
updatedAt: event.payload.updatedAt,
});
return;
}

case "thread.meta-updated": {
const existingRow = yield* projectionThreadRepository.getById({
threadId: event.payload.threadId,
Expand Down Expand Up @@ -2183,6 +2254,7 @@ export const OrchestrationProjectionPipelineLive = Layer.effect(
makeOrchestrationProjectionPipeline(),
).pipe(
Layer.provideMerge(ProjectionProjectRepositoryLive),
Layer.provideMerge(ProjectionThreadGroups.layer),
Layer.provideMerge(ProjectionThreadRepositoryLive),
Layer.provideMerge(ProjectionThreadMessageRepositoryLive),
Layer.provideMerge(ProjectionThreadProposedPlanRepositoryLive),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -487,6 +487,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
activeOrderKey: "hq",
titleRegeneration: null,
titleState: null,
groupId: null,
deletedAt: null,
messages: [
{
Expand Down Expand Up @@ -613,6 +614,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
activeOrderKey: "hq",
titleRegeneration: null,
titleState: null,
groupId: null,
session: {
threadId: ThreadId.make("thread-1"),
status: "running",
Expand Down
Loading
Loading