Skip to content
26 changes: 23 additions & 3 deletions apps/server/src/mcp/OrchestratorMcpService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3231,17 +3231,37 @@ describe("OrchestratorMcpService provider resolution", () => {
}),
);

it.effect("keeps the saved workspace when an unbound task stays unbound", () =>
it.effect.each([
{
name: "keeps a project folder chosen in Settings",
task: { ...existing, creationSource: "web" },
workspace: { type: "root" },
},
{
name: "keeps a saved worktree",
task: {
...existing,
workspaceStrategy: { type: "worktree", baseRef: "develop", startFromOrigin: false },
},
workspace: { type: "worktree", baseRef: "develop", startFromOrigin: false },
},
{
// #69: saved as root while the folder was not a repository; it is one now.
name: "re-derives a project folder it saved, now that the folder is a repository",
task: existing,
workspace: { type: "worktree", baseRef: "origin/master", startFromOrigin: true },
},
] as const)("an unbound task that stays unbound $name", ({ task, workspace }) =>
Effect.gen(function* () {
const saved = yield* Ref.make<ReadonlyArray<ScheduledTask["workspaceStrategy"]>>([]);
const mcp = yield* OrchestratorMcpService.OrchestratorMcpService.pipe(
Effect.provide(service(masterRepository, saved)),
Effect.provide(service(masterRepository, saved, task)),
);
yield* mcp.updateScheduledTask(client, {
scheduledTaskId: existing.id,
bindToCurrentThread: false,
});
assert.deepEqual(yield* Ref.get(saved), [{ type: "root" }]);
assert.deepEqual(yield* Ref.get(saved), [workspace]);
}),
);

Expand Down
13 changes: 10 additions & 3 deletions apps/server/src/mcp/OrchestratorMcpService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ import {
type OrchestratorMcpDeleteScheduledTaskResult,
type OrchestratorMcpListScheduledTasksResult,
type OrchestratorMcpScheduledTask,
orchestratorMcpScheduledTaskRunStatus,
type OrchestratorMcpScheduleTaskInput,
type OrchestratorMcpScheduleTaskResult,
type OrchestratorMcpTarget,
Expand Down Expand Up @@ -293,7 +294,7 @@ function scheduledTaskSummary(task: ScheduledTask, mayRun: boolean): Orchestrato
freshSession: scheduledTaskThreadMode(task) === "fresh_session",
schedule: task.schedule,
nextRunAt: task.nextRunAt,
lastRunStatus: task.lastRunStatus,
lastRunStatus: orchestratorMcpScheduledTaskRunStatus(task),
...(task.webhook?.url == null || !mayRun || task.blockedByPolicy !== undefined
? {}
: { webhookUrl: task.webhook.url }),
Expand Down Expand Up @@ -2169,9 +2170,15 @@ const make = Effect.gen(function* () {
: input.bindToCurrentThread && parent !== undefined
? parent.thread.id
: null;
// Reset the workspace only when the binding flips, so unbinding cannot leave runs in the shared checkout.
// Reset the workspace when the binding flips, so unbinding cannot leave runs in the shared
// checkout. An unbound root it derived is re-derived too: the folder may be a repository now.
const workspaceStrategy =
(threadId === null) === (existing.threadId === null)
(threadId === null) === (existing.threadId === null) &&
!(
threadId === null &&
existing.creationSource === "mcp" &&
existing.workspaceStrategy.type === "root"
)
? existing.workspaceStrategy
: yield* scheduledTaskWorkspaceStrategy(threadId !== null, existing.projectId);
const upsertInput: ScheduledTaskUpsertInput = {
Expand Down
29 changes: 21 additions & 8 deletions apps/server/src/mcp/gitEntryProbe.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,20 +3,33 @@ import * as NodeFS from "node:fs";
import * as NodePath from "node:path";
import * as Effect from "effect/Effect";

export interface GitEntry {
readonly path: string;
/** The folder holding it, which is `cwd` itself or one of its parents. */
readonly folder: string;
readonly inParent: boolean;
}

/**
* Whether `cwd` or any parent folder has a `.git` entry (folder or gitfile). A folder that cannot be
* inspected counts as having one, so callers that use this to prove "no repository" fail closed.
* The nearest `.git` entry (folder or gitfile) in `cwd` or any parent folder, or null. A folder that
* cannot be inspected counts as having one, so callers that use this to prove "no repository" fail
* closed.
*/
export const hasGitEntry = (cwd: string): Effect.Effect<boolean> =>
export const findGitEntry = (cwd: string): Effect.Effect<GitEntry | null> =>
Effect.sync(() => {
for (let folder = NodePath.resolve(cwd); ; folder = NodePath.dirname(folder)) {
const start = NodePath.resolve(cwd);
for (let folder = start; ; folder = NodePath.dirname(folder)) {
const found = { path: NodePath.join(folder, ".git"), folder, inParent: folder !== start };
try {
NodeFS.lstatSync(NodePath.join(folder, ".git"));
return true;
NodeFS.lstatSync(found.path);
return found;
} catch (error) {
const code = (error as NodeJS.ErrnoException).code;
if (code !== "ENOENT" && code !== "ENOTDIR") return true;
if (code !== "ENOENT" && code !== "ENOTDIR") return found;
}
if (NodePath.dirname(folder) === folder) return false;
if (NodePath.dirname(folder) === folder) return null;
}
});

export const hasGitEntry = (cwd: string): Effect.Effect<boolean> =>
Effect.map(findGitEntry(cwd), (entry) => entry !== null);
3 changes: 2 additions & 1 deletion apps/server/src/mcp/toolkits/thread/handlers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import {
type OrchestrationV2ThreadProjection,
type RunId,
OrchestratorMcpFailure,
orchestratorMcpScheduledTaskRunStatus,
type OrchestrationV2Command,
} from "@t3tools/contracts";
import * as Effect from "effect/Effect";
Expand Down Expand Up @@ -130,7 +131,7 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, {
: {
link: McpInvocationContext.threadLinkFor(scope, threadId),
}),
lastRunStatus: task.lastRunStatus,
lastRunStatus: orchestratorMcpScheduledTaskRunStatus(task),
runCount: task.runCount,
nextRunAt: task.nextRunAt,
};
Expand Down
3 changes: 2 additions & 1 deletion apps/server/src/mcp/toolkits/thread/tools.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import {
ProviderUserInputAnswers,
IsoDateTime,
OrchestratorMcpFailure,
OrchestratorMcpScheduledTaskRunStatus,
OrchestrationV2DispatchCommandResult,
ThreadId,
RunId,
Expand Down Expand Up @@ -268,7 +269,7 @@ const ScheduledTaskRunTool = Tool.make("run_scheduled_task_now", {
taskId: ScheduledTaskId,
threadId: ScheduledTask.fields.threadId,
link: Schema.optional(Schema.String),
lastRunStatus: ScheduledTask.fields.lastRunStatus,
lastRunStatus: OrchestratorMcpScheduledTaskRunStatus,
runCount: NonNegativeInt,
nextRunAt: ScheduledTask.fields.nextRunAt,
}),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -326,7 +326,7 @@ it.effect.each(deliveries)(
}),
);
const ran = yield* service.runNow({ id: task.id });
assert.equal(ran.task.lastRunStatus, "succeeded");
assert.equal(ran.task.runInProgressSince, ran.task.lastRunAt);
}).pipe(
Effect.provide(
ScheduledTaskService.layer.pipe(
Expand Down
6 changes: 3 additions & 3 deletions apps/server/src/orchestration-v2/ThreadLaunchService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -377,7 +377,7 @@ it.effect.each(
creationSource: createdBy === "agent" ? "mcp" : "web",
});
const result = yield* tasks.runNow({ id: task.id });
assert.equal(result.task.lastRunStatus, "succeeded");
assert.equal(result.task.runInProgressSince, result.task.lastRunAt);
const projectThreads = yield* threads.listProjectThreads({
projectId,
includeSubagents: false,
Expand Down Expand Up @@ -494,7 +494,7 @@ it.effect("runs a thread-bound automation on the provider the thread uses now",
});

const result = yield* tasks.runNow({ id: task.id });
assert.equal(result.task.lastRunStatus, "succeeded");
assert.equal(result.task.runInProgressSince, result.task.lastRunAt);
const projection = yield* threads.getThreadProjection(launched.threadId);
const message = projection.messages.find((item) => item.scheduledTaskId === task.id);
const run = projection.runs.find((item) => item.id === message?.runId);
Expand Down Expand Up @@ -1086,7 +1086,7 @@ it.effect("runs a Scratch scheduled task saved with a worktree in its own folder
WHERE task_id = ${task.id}
`;
const ran = yield* tasks.runNow({ id: task.id });
assert.equal(ran.task.lastRunStatus, "succeeded");
assert.equal(ran.task.runInProgressSince, ran.task.lastRunAt);
const [thread] = yield* threads.listProjectThreads({ projectId, includeSubagents: false });
assert.isDefined(thread);
const projection = yield* preparationSettled(thread!.id);
Expand Down
129 changes: 128 additions & 1 deletion apps/server/src/scheduledTasks/ScheduledTaskService.schedule.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,14 +6,18 @@ import {
type OrchestrationV2Run,
MessageId,
RunId,
type ScheduledTask,
ScheduledTaskUpsertInput,
} from "@t3tools/contracts";
import * as DateTime from "effect/DateTime";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Queue from "effect/Queue";
import * as Ref from "effect/Ref";
import * as Schema from "effect/Schema";
import * as Stream from "effect/Stream";
import * as SqlClient from "effect/sql/SqlClient";
import * as TestClock from "effect/testing/TestClock";

import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
Expand Down Expand Up @@ -466,6 +470,128 @@ it.effect("a fixed-time slot missed while the server was off shows on the task",
}).pipe(Effect.provide(SqlitePersistence.layerMemory)),
);

it.effect(
"a slot missed during a restart is not overwritten by the run still going from before",
() =>
Effect.gen(function* () {
const at = (hour: number, minute = 0) =>
DateTime.makeZonedUnsafe(
{ year: 2026, month: 7, day: 1, hour, minute, second: 0, millisecond: 0 },
{ timeZone: DateTime.zoneMakeLocal(), adjustForTimeZone: true },
);
yield* TestClock.setTime(DateTime.toEpochMillis(at(8)));
const sql = yield* SqlClient.SqlClient;
const sent = yield* Ref.make("");
const end = yield* Deferred.make<void>();
const threadId = "thread-pending-run";
const layer = ScheduledTaskService.layer.pipe(
Layer.provide(
Layer.mergeAll(
NodeCrypto.layer,
Scheduler.layer,
ManagedPolicy.layerTest(),
Layer.mock(ThreadLaunchService.ThreadLaunchService)({
launch: (input) =>
Ref.set(sent, input.initialMessage?.messageId ?? "").pipe(
Effect.as({ threadId } as never),
),
}),
Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
namedProjectsRoot: "/projects",
isScratchProject: () => Effect.succeed(false),
}),
Layer.mock(ThreadManagementService.ThreadManagementService)({
getThreadShell: () =>
Effect.succeed({ projectId: "project-pending-run", deletedAt: null } as never),
getProjectThreadRecords: () =>
Effect.map(
Ref.get(sent),
(userMessageId) =>
({
runs: [{ id: "run:1", status: "running", userMessageId }],
messages: [{ id: userMessageId, runId: "run:1" }],
}) as never,
),
waitForThread: (input) =>
Deferred.await(end).pipe(
Effect.as({
threadId: input.threadId,
run: { id: "run:1", status: "completed", rootNodeId: "node:root" },
timedOut: false,
} as never),
),
}),
Layer.mock(SecretRequests.SecretRequests)({}),
),
),
);

const taskId = yield* Effect.gen(function* () {
const service = yield* ScheduledTaskService.ScheduledTaskService;
const { task } = yield* service.upsert(
yield* decodeUpsertInput({
title: "Morning bookkeeping",
prompt: "Update yesterday's notes.",
enabled: true,
schedule: { type: "fixed_time", timeOfDay: "09:00" },
projectId: "project-pending-run",
workspaceStrategy: { type: "root" },
modelSelection: { instanceId: "codex", model: "gpt-5.4" },
runtimeMode: "full-access",
interactionMode: "default",
}),
);
yield* service.runNow({ id: task.id });
return task.id;
}).pipe(Effect.provide(layer));
// What the launch committed; the mocked launch writes no projection.
yield* sql`
INSERT INTO orchestration_v2_projection_messages
(message_id, thread_id, run_id, role, streaming, created_at, updated_at, payload_json)
VALUES (${yield* Ref.get(sent)}, ${threadId}, 'run:1', 'user', 0, '', '', '{}')
`;

// The server is off across the 09:00 slot while the 08:00 run keeps going.
yield* TestClock.setTime(DateTime.toEpochMillis(at(9, 30)));
const missedError = yield* Effect.gen(function* () {
const service = yield* ScheduledTaskService.ScheduledTaskService;
const states = yield* Queue.unbounded<ScheduledTask | undefined>();
yield* service.subscribeList().pipe(
Stream.runForEach(({ tasks }) =>
Queue.offer(
states,
tasks.find((candidate) => candidate.id === taskId),
),
),
Effect.forkChild,
);
expect((yield* Queue.take(states))?.runInProgressSince).toBeDefined();
yield* TestClock.adjust("6 seconds");
const missed = yield* Queue.take(states);
expect(missed?.lastRunStatus).toBe("failed");
yield* Deferred.succeed(end, undefined);
const afterRun = yield* Queue.take(states);
expect({
status: afterRun?.lastRunStatus,
error: afterRun?.lastRunError,
since: afterRun?.runInProgressSince,
}).toEqual({ status: "failed", error: missed?.lastRunError, since: undefined });
return missed?.lastRunError;
}).pipe(Effect.provide(layer));

// A second restart must not find the 08:00 run again and record its success over the miss.
const [again] = yield* Effect.gen(function* () {
const service = yield* ScheduledTaskService.ScheduledTaskService;
return (yield* service.list()).tasks;
}).pipe(Effect.provide(layer));
expect({
status: again?.lastRunStatus,
error: again?.lastRunError,
since: again?.runInProgressSince,
}).toEqual({ status: "failed", error: missedError, since: undefined });
}).pipe(Effect.provide(SqlitePersistence.layerMemory)),
);

it.effect(
"a run made while the scheduler was busy is not overwritten by its stale missed slot",
() =>
Expand Down Expand Up @@ -549,8 +675,9 @@ it.effect(
}),
);

// The manual run's outcome is still pending, and the stale slot recorded no miss over it.
expect({ status: morning?.lastRunStatus, error: morning?.lastRunError }).toEqual({
status: "succeeded",
status: "never",
error: null,
});
expect(morning?.nextRunAt).toBe(DateTime.formatIso(DateTime.toUtc(at(8, 0, 2))));
Expand Down
Loading
Loading