Skip to content
Draft
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
2 changes: 2 additions & 0 deletions apps/server/src/mcp/toolkits/environment/handlers.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,8 @@ const layerEnvironment = Layer.mock(Environment.ServerEnvironment)({

it.effect("updates allowed MCP preferences without enabling automatic continuity", () =>
Effect.gen(function* () {
const seedSettings = yield* Settings.ServerSettingsService;
yield* seedSettings.updateSettings({ workContinuityAutoEnroll: false });
const server = yield* McpServer.McpServer;
const result = yield* server
.callTool({
Expand Down
8 changes: 8 additions & 0 deletions apps/server/src/orchestration-v2/ThreadManagementService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -925,6 +925,14 @@ const make = Effect.gen(function* () {
)(function* (input) {
const shell = input.byOwnAgent ? yield* orchestrator.getThreadShell(input.threadId) : null;
if (shell != null && shell.activeRunId !== null) {
// A pending own-agent settle is a deliberate end: supervision must not
// spend a continuation check on the turn that is about to settle.
yield* dispatch({
type: "thread.continuity.set",
commandId: CommandId.make(`${input.commandId}:continuity-off`),
threadId: input.threadId,
enabled: false,
}).pipe(Effect.ignoreCause({ log: true }));
yield* settleAfterRun({
projectId: shell.projectId,
threadId: shell.id,
Expand Down
75 changes: 61 additions & 14 deletions apps/server/src/orchestration-v2/runtimeLayer.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -844,6 +844,8 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => {

it.effect("resolves delivery intent against the active run and starts after it completes", () =>
Effect.gen(function* () {
const settings = yield* ServerSettings.ServerSettingsService;
yield* settings.updateSettings({ workContinuityAutoEnroll: false });
const orchestrator = yield* Orchestrator.OrchestratorV2;
const eventSink = yield* EventSink.EventSinkV2;
const outbox = yield* EffectOutbox.EffectOutboxV2;
Expand Down Expand Up @@ -3256,6 +3258,8 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => {
"does not admit a restart continuation of a %s run with only lost background work",
(status) =>
Effect.gen(function* () {
const settings = yield* ServerSettings.ServerSettingsService;
yield* settings.updateSettings({ workContinuityAutoEnroll: false });
const orchestrator = yield* Orchestrator.OrchestratorV2;
const eventSink = yield* EventSink.EventSinkV2;
const threadId = ThreadId.make(`runtime-layer-restart-${status}-source`);
Expand Down Expand Up @@ -5725,7 +5729,17 @@ it.layer(layerTest.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))(
yield* settings.updateSettings({ workContinuityAutoEnroll: true });
return yield* program.pipe(
Effect.ensuring(
settings.updateSettings({ workContinuityAutoEnroll: false }).pipe(Effect.orDie),
settings.updateSettings({ workContinuityAutoEnroll: true }).pipe(Effect.orDie),
),
);
});
const withoutAutoEnrollment = <A, E, R>(program: Effect.Effect<A, E, R>) =>
Effect.gen(function* () {
const settings = yield* ServerSettings.ServerSettingsService;
yield* settings.updateSettings({ workContinuityAutoEnroll: false });
return yield* program.pipe(
Effect.ensuring(
settings.updateSettings({ workContinuityAutoEnroll: true }).pipe(Effect.orDie),
),
);
});
Expand All @@ -5745,13 +5759,42 @@ it.layer(layerTest.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))(
);

it.effect("disabled auto enrollment leaves accepted human messages unenrolled", () =>
Effect.gen(function* () {
const input = yield* seedContinuity("auto-off", undefined, true);
const orchestrator = yield* Orchestrator.OrchestratorV2;
assert.isNull(
(yield* orchestrator.getThreadProjection(input.threadId)).thread.continuity ?? null,
);
}),
withoutAutoEnrollment(
Effect.gen(function* () {
const input = yield* seedContinuity("auto-off", undefined, true);
const orchestrator = yield* Orchestrator.OrchestratorV2;
assert.isNull(
(yield* orchestrator.getThreadProjection(input.threadId)).thread.continuity ?? null,
);
}),
),
);

it.effect(
"auto enrollment is on by default and spends exactly one check after a missing terminal report",
() =>
Effect.gen(function* () {
const input = yield* seedContinuity("auto-default", undefined, true);
const orchestrator = yield* Orchestrator.OrchestratorV2;
const before = (yield* orchestrator.getThreadProjection(input.threadId)).thread
.continuity;
assert.equal(before?.authorityMessageId, input.authorityMessageId);
assert.equal(before?.checksRemaining, 3);
const events = yield* EventSink.EventSinkV2;
const started = yield* events
.stream({ afterSequence: yield* events.latestSequence(), eventType: "run.created" })
.pipe(
Stream.filter((stored) => stored.event.threadId === input.threadId),
Stream.take(1),
Stream.runCollect,
Effect.forkChild({ startImmediately: true }),
);
yield* finishRun(input.threadId, input.run);
yield* Fiber.join(started);
const after = yield* orchestrator.getThreadProjection(input.threadId);
assert.lengthOf(after.runs, 2);
assert.equal(after.thread.continuity?.checksRemaining, 2);
}),
);

it.effect(
Expand Down Expand Up @@ -5803,7 +5846,7 @@ it.layer(layerTest.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))(
"enabling the setting alone does not enroll or wake existing work, but a queued human turn does",
() =>
Effect.gen(function* () {
const input = yield* seedContinuity("auto-queued", undefined, true);
const input = yield* withoutAutoEnrollment(seedContinuity("auto-queued", undefined, true));
const orchestrator = yield* Orchestrator.OrchestratorV2;
yield* withAutoEnrollment(
Effect.gen(function* () {
Expand Down Expand Up @@ -5962,7 +6005,7 @@ it.layer(layerTest.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))(
"cancelling the queued authority blocks supervision without spending or replacing its allowance",
() =>
Effect.gen(function* () {
const input = yield* seedContinuity("auto-queued-cancel", undefined, true);
const input = yield* withoutAutoEnrollment(seedContinuity("auto-queued-cancel", undefined, true));
const orchestrator = yield* Orchestrator.OrchestratorV2;
yield* withAutoEnrollment(
Effect.gen(function* () {
Expand Down Expand Up @@ -6190,6 +6233,7 @@ it.layer(layerTest.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))(
it.effect.each(["server", "mcp", "provider"] as const)(
"human-shaped %s dispatch does not automatically enroll",
(source) =>
withoutAutoEnrollment(
Effect.gen(function* () {
const input = yield* seedContinuity(`auto-source-${source}`, undefined, true);
const orchestrator = yield* Orchestrator.OrchestratorV2;
Expand All @@ -6211,6 +6255,7 @@ it.layer(layerTest.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))(
(yield* orchestrator.getThreadProjection(input.threadId)).thread.continuity ?? null,
);
}),
),
);
it.effect.each([
"plan",
Expand All @@ -6221,6 +6266,7 @@ it.layer(layerTest.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))(
"scheduled",
"sender",
] as const)("%s prevents automatic continuity enrollment", (exclusion) =>
withoutAutoEnrollment(
Effect.gen(function* () {
const input = yield* seedContinuity(`auto-excluded-${exclusion}`, undefined, true);
const orchestrator = yield* Orchestrator.OrchestratorV2;
Expand Down Expand Up @@ -6296,6 +6342,7 @@ it.layer(layerTest.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))(
(yield* orchestrator.getThreadProjection(input.threadId)).thread.continuity ?? null,
);
}),
),
);

it.effect(
Expand Down Expand Up @@ -6479,10 +6526,10 @@ it.layer(layerTest.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))(
})
.pipe(Effect.flip);
assert.instanceOf(refused, Orchestrator.OrchestratorDispatchError);
assert.equal(
(yield* orchestrator.getThreadProjection(input.threadId)).thread.continuity?.status,
"complete",
);
const reenrolled = (yield* orchestrator.getThreadProjection(input.threadId)).thread
.continuity;
assert.equal(reenrolled?.status, "active");
assert.equal(reenrolled?.authorityMessageId, MessageId.make(`${input.threadId}:new-request`));
}),
);

Expand Down
4 changes: 3 additions & 1 deletion apps/server/src/ws.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1846,7 +1846,9 @@ const layerWsRpc = (
ThreadManagementService.withCreationProvenance(command, {
createdBy: "user",
creationSource:
"creationSource" in command ? command.creationSource : "web",
"creationSource" in command
? (command.creationSource ?? "web")
: "web",
}),
)
).pipe(Effect.provide(intakeContext)),
Expand Down
4 changes: 2 additions & 2 deletions packages/contracts/src/settings.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -24,8 +24,8 @@ const encodeServerSettings = Schema.encodeSync(ServerSettings);
const decodeClaudeSettings = Schema.decodeUnknownSync(ClaudeSettings);

describe("work continuity auto enrollment", () => {
it("keeps automatic enrollment disabled when decoding old settings", () => {
expect(decodeServerSettings({}).workContinuityAutoEnroll).toBe(false);
it("enables automatic enrollment when decoding old settings without the key", () => {
expect(decodeServerSettings({}).workContinuityAutoEnroll).toBe(true);
});
it("round-trips the environment enrollment toggle without changing other settings", () => {
expect(decodeServerSettingsPatch({ workContinuityAutoEnroll: true })).toEqual({
Expand Down
2 changes: 1 addition & 1 deletion packages/contracts/src/settings.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1081,7 +1081,7 @@ export const StorageCleanupSettings = Schema.Struct({
export type StorageCleanupSettings = typeof StorageCleanupSettings.Type;

export const ServerSettings = Schema.Struct({
workContinuityAutoEnroll: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))),
workContinuityAutoEnroll: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))),
// Null means no client has migrated this preference yet; an empty value is
// authoritative so another device cannot resurrect cleared preferences.
favorites: Schema.NullOr(ModelFavorites).pipe(Schema.withDecodingDefault(Effect.succeed(null))),
Expand Down
Loading