Skip to content
Merged
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
68 changes: 66 additions & 2 deletions apps/server/src/orchestration/Layers/ProviderCommandReactor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import { isTemporaryWorktreeBranch, WORKTREE_BRANCH_PREFIX } from "@t3tools/shar
import * as Cache from "effect/Cache";
import * as Cause from "effect/Cause";
import * as Crypto from "effect/Crypto";
import * as DateTime from "effect/DateTime";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Equal from "effect/Equal";
Expand Down Expand Up @@ -52,7 +53,8 @@ type ProviderIntentEvent = Extract<
| "thread.turn-interrupt-requested"
| "thread.approval-response-requested"
| "thread.user-input-response-requested"
| "thread.session-stop-requested";
| "thread.session-stop-requested"
| "thread.context-compacted";
}
>;

Expand Down Expand Up @@ -944,6 +946,64 @@ const make = Effect.gen(function* () {
});
});

const processContextCompacted = Effect.fn("processContextCompacted")(function* (
event: Extract<ProviderIntentEvent, { type: "thread.context-compacted" }>,
) {
const threadId = event.payload.threadId;
const thread = yield* resolveThread(threadId);
if (!thread) {
return;
}

const nowIso = yield* Effect.map(DateTime.now, DateTime.formatIso);

// Attempt provider compaction
const compactResult = yield* providerService.compactThread({ threadId }).pipe(
Effect.map((result) => ({ _tag: "success" as const, ...result })),
Effect.catchAll((cause) =>
Effect.gen(function* () {
yield* Effect.logWarning("provider compaction failed, proceeding with trim-only", {
threadId: String(threadId),
cause: Cause.pretty(cause),
});
return { _tag: "failure" as const };
}),
),
);

if (compactResult._tag === "success") {
// Dispatch summarize with the compaction summary
const summarizeCommandId = yield* serverCommandId("compact-summarize");
yield* orchestrationEngine.dispatch({
type: "thread.context.summarize",
commandId: summarizeCommandId,
threadId,
summary: compactResult.summary,
compactDurationMs: compactResult.durationMs,
createdAt: nowIso,
});

// Dispatch trim with the summary attached
const trimCommandId = yield* serverCommandId("compact-trim");
yield* orchestrationEngine.dispatch({
type: "thread.context.trim",
commandId: trimCommandId,
threadId,
summary: compactResult.summary,
createdAt: nowIso,
Comment on lines +988 to +993

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Avoid pruning turns created while compacting

After provider compaction finishes, this dispatch trims the current read model with no boundary or keepLastNTurns, while the earlier compact intent does not mark the thread busy. If a user submits and completes a turn during a long compaction, the provider summary was started before that turn but this later trim can still prune the new messages from the UI/read model; record the pre-compaction boundary or block/preserve turns created after compaction starts.

Useful? React with 👍 / 👎.

});
} else {
// Dispatch trim without summary (fallback)
const trimCommandId = yield* serverCommandId("compact-trim-fallback");
yield* orchestrationEngine.dispatch({
type: "thread.context.trim",
commandId: trimCommandId,
threadId,
createdAt: nowIso,
});
}
});

const processDomainEvent = Effect.fn("processDomainEvent")(function* (
event: ProviderIntentEvent,
) {
Expand Down Expand Up @@ -984,6 +1044,9 @@ const make = Effect.gen(function* () {
case "thread.session-stop-requested":
yield* processSessionStopRequested(event);
return;
case "thread.context-compacted":
yield* processContextCompacted(event);
return;
}
});

Expand All @@ -1010,7 +1073,8 @@ const make = Effect.gen(function* () {
event.type === "thread.turn-interrupt-requested" ||
event.type === "thread.approval-response-requested" ||
event.type === "thread.user-input-response-requested" ||
event.type === "thread.session-stop-requested"
event.type === "thread.session-stop-requested" ||
event.type === "thread.context-compacted"
) {
return yield* worker.enqueue(event);
}
Expand Down
Loading
Loading