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
Original file line number Diff line number Diff line change
Expand Up @@ -89,8 +89,10 @@ import {
import { appAtomRegistry } from "../../state/atom-registry";
import {
formatNextScheduledTaskRun,
formatScheduledTaskArchive,
formatScheduledTaskInterval,
} from "./scheduledTaskPresentation";
import { scheduledTaskRestoreNeedsEditor } from "@t3tools/client-runtime/scheduled-task-archive";

type ScheduledTaskRoutes = {
SettingsScheduledTaskNew: undefined;
Expand Down Expand Up @@ -414,6 +416,7 @@ export function SettingsScheduledTasksRouteScreen() {
const navigation = useNavigation<NativeStackNavigationProp<ScheduledTaskRoutes>>();
const insets = useSafeAreaInsets();
const environmentMenu = useSettingsEnvironmentMenu();
const [view, setView] = useState<"active" | "archived">("active");
const newTask = () => {
if (visibleEnvironments.length === 0) return;
resetEditor();
Expand Down Expand Up @@ -442,10 +445,19 @@ export function SettingsScheduledTasksRouteScreen() {
contentContainerClassName="gap-5 px-5 pt-4"
contentContainerStyle={{ paddingBottom: Math.max(insets.bottom, 18) + 18 }}
>
<SegmentedControl
options={[
{ value: "active" as const, label: "Active" },
{ value: "archived" as const, label: "Archived" },
]}
selected={view}
onSelect={setView}
/>
{visibleEnvironments.length > 0 ? (
visibleEnvironments.map((environment) => (
<EnvironmentTasks
key={environment.environmentId}
view={view}
environment={environment}
now={now}
projectIds={
Expand Down Expand Up @@ -1397,11 +1409,13 @@ function WebhookScheduleDetails({
}

function EnvironmentTasks({
view,
environment,
now,
projectIds,
onEdit,
}: {
readonly view: "active" | "archived";
readonly environment: SettingsTarget;
readonly now: number;
readonly projectIds: readonly ProjectId[] | null;
Expand All @@ -1412,7 +1426,9 @@ function EnvironmentTasks({
serverEnvironment.scheduledTasksLive({ environmentId, input: {} }),
);
const visibleTasks = tasks.data?.tasks.filter(
(task) => projectIds === null || projectIds.includes(task.projectId),
(task) =>
(projectIds === null || projectIds.includes(task.projectId)) &&
(task.archivedAt != null) === (view === "archived"),
);
const canOperate = useAtomValue(
serverEnvironment.upsertScheduledTask.permissionAtom(environmentId),
Expand All @@ -1436,15 +1452,22 @@ function EnvironmentTasks({
}
};

const act = async (task: ScheduledTask, action: "run" | "toggle" | "delete") => {
const act = async (task: ScheduledTask, action: "run" | "toggle" | "restore" | "delete") => {
if (!readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) return;
if (action === "restore" && scheduledTaskRestoreNeedsEditor(task, now)) {
onEdit(task);
return;
}
const result =
action === "run"
? await runNow({ environmentId, input: { id: task.id } })
: action === "toggle"
: action === "toggle" || action === "restore"
? await setEnabled({ environmentId, input: { id: task.id, enabled: !task.enabled } })
: await remove({ environmentId, input: { id: task.id } });
failure(`Could not ${action === "toggle" ? "update" : action} task`, result);
failure(
`Could not ${action === "toggle" || action === "restore" ? "update" : action} task`,
result,
);
};

return (
Expand All @@ -1466,7 +1489,11 @@ function EnvironmentTasks({
<Text className="p-4 text-base text-foreground-muted">Loading tasks…</Text>
) : visibleTasks?.length === 0 ? (
<Text className="p-4 text-base text-foreground-muted">
{projectIds === null ? "No scheduled tasks yet." : "No tasks in this project."}
{view === "archived"
? "No archived tasks."
: projectIds === null
? "No scheduled tasks yet."
: "No tasks in this project."}
</Text>
) : (
visibleTasks?.map((task, index) => (
Expand All @@ -1491,11 +1518,13 @@ function EnvironmentTasks({
</Text>
<Text className="text-sm text-foreground-muted" numberOfLines={2}>
{describeSchedule(task, tasks.data?.timeZone)}
{!task.enabled
? " · Paused"
: task.nextRunAt
? ` · ${formatNextScheduledTaskRun(task.nextRunAt, now)}`
: ""}
{task.archivedAt != null
? ` · ${formatScheduledTaskArchive(task)}`
: !task.enabled
? " · Paused"
: task.nextRunAt
? ` · ${formatNextScheduledTaskRun(task.nextRunAt, now)}`
: ""}
</Text>
{task.blockedByPolicy ? (
<Text className="text-sm text-foreground-muted" numberOfLines={2}>
Expand All @@ -1514,12 +1543,14 @@ function EnvironmentTasks({
...(task.threadId !== null && scheduledTaskThreadMode(task) !== "new_thread"
? [{ id: "thread", title: "Open thread" }]
: []),
{
id: "toggle",
title: task.enabled ? "Pause" : "Resume",
attributes: { disabled: !canOperate },
},
...(task.schedule.type === "webhook"
task.archivedAt != null
? { id: "restore", title: "Restore", attributes: { disabled: !canOperate } }
: {
id: "toggle",
title: task.enabled ? "Pause" : "Resume",
attributes: { disabled: !canOperate },
},
...(task.schedule.type === "webhook" || task.archivedAt != null
? []
: [{ id: "run", title: "Run now", attributes: { disabled: !canOperate } }]),
{
Expand Down Expand Up @@ -1549,7 +1580,7 @@ function EnvironmentTasks({
onPress: () => void act(task, "delete"),
},
]);
} else if (action === "toggle" || action === "run") {
} else if (action === "toggle" || action === "restore" || action === "run") {
void act(task, action);
}
}}
Expand Down
3 changes: 2 additions & 1 deletion apps/mobile/src/features/settings/scheduledTaskDraft.ts
Original file line number Diff line number Diff line change
Expand Up @@ -322,7 +322,8 @@ export function editDraft(task: ScheduledTask): ScheduledTaskDraft {
task.workspaceStrategy.type === "existing_worktree"
? task.workspaceStrategy.worktreePath
: "",
enabled: task.enabled,
// Saving restores an archived task, so it opens ready to run again.
enabled: task.enabled || task.archivedAt != null,
startFromOrigin:
task.workspaceStrategy.type === "worktree"
? (task.workspaceStrategy.startFromOrigin ?? false)
Expand Down
13 changes: 13 additions & 0 deletions apps/mobile/src/features/settings/scheduledTaskPresentation.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
import { scheduledTaskArchiveLabel } from "@t3tools/client-runtime/scheduled-task-archive";
import { scheduledTaskDeletesAt, type ScheduledTask } from "@t3tools/contracts";

const MINUTE = 60_000;
const HOUR = 60 * MINUTE;
const DAY = 24 * HOUR;
Expand Down Expand Up @@ -55,3 +58,13 @@ export function formatNextScheduledTaskRun(nextRunAt: string, now: number): stri
});
return `Next run ${date} at ${time}`;
}

export function formatScheduledTaskArchive(
task: Pick<ScheduledTask, "archivedAt" | "archiveReason">,
): string {
const label = scheduledTaskArchiveLabel(task);
const deletesAt = scheduledTaskDeletesAt(task);
if (deletesAt === null) return label;
const date = new Date(deletesAt).toLocaleDateString([], { dateStyle: "medium" });
return `${label} · Deletes on ${date}`;
}
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,7 @@ const layerUnusedScheduledTaskStub = Layer.succeed(
getWebhookDelivery: () => Effect.die("unused in this test"),
triggerWebhook: () => Effect.die("unused in this test"),
awaitWebhookDeliveries: Effect.void,
cleanupStale: () => Effect.die("unused in this test"),
}),
);

Expand Down
31 changes: 30 additions & 1 deletion apps/server/src/orchestration-v2/ProjectionSettlement.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import {
import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as SqlClient from "effect/sql/SqlClient";

import * as SqlitePersistence from "../persistence/Sqlite.ts";
import * as ProjectionStore from "./ProjectionStore.ts";
Expand Down Expand Up @@ -531,12 +532,40 @@ it.effect("sql: lists threads settled before the cutoff for auto-archive, oldest
lineage: { rootThreadId: parent, parentThreadId: parent, relationshipToParent: "subagent" },
});
yield* createRun(subagent, "completed");
// A schedule that still runs into the thread keeps it; a paused one does not.
const sql = yield* SqlClient.SqlClient;
const schedule = (taskId: string, threadId: string, enabled: number) =>
sql`INSERT INTO scheduled_tasks ${sql.insert({
task_id: taskId,
title: taskId,
prompt: "Check in",
enabled,
schedule_json: '{"type":"cron","expression":"0 9 1 * *","timezone":"UTC"}',
project_id: "project-1",
thread_id: threadId,
workspace_strategy_json: '{"type":"root"}',
model_selection_json: '{"instanceId":"codex","model":"gpt-5"}',
runtime_mode: "full-access",
interaction_mode: "default",
created_by: "agent",
creation_source: "mcp",
created_at: DateTime.formatIso(old),
updated_at: DateTime.formatIso(old),
next_run_at: null,
last_run_at: null,
last_run_status: "never",
last_run_error: null,
run_count: 0,
})}`;
yield* schedule("monthly", yield* createThread("archive-scheduled", settled(30)), 1);
const pausedSchedule = yield* createThread("archive-paused-schedule", settled(25));
yield* schedule("paused", pausedSchedule, 0);

const settledBefore = DateTime.subtract(now, { days: 10 });
const candidates = yield* store.getAutoArchiveCandidates({ settledBefore, limit: 10 });
assert.deepEqual(
candidates.map((thread) => thread.id),
[oldest, parent, child, due],
[oldest, pausedSchedule, parent, child, due],
);
assert.deepEqual(
(yield* store.getAutoArchiveCandidates({ settledBefore, limit: 1 })).map(
Expand Down
5 changes: 5 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7207,6 +7207,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
AND COALESCE(json_extract(t.payload_json, '$.lineage.relationshipToParent'), '') != 'subagent'
AND COALESCE(json_extract(t.payload_json, '$.lineage.sideChat'), 0) = 0
AND ${noBlockingWork}
-- Archiving would make every later run of a task bound here fail.
AND NOT EXISTS (
SELECT 1 FROM scheduled_tasks task
WHERE task.thread_id = t.thread_id AND task.enabled = 1
)
AND NOT EXISTS (
SELECT 1 FROM orchestration_v2_projection_threads child
-- Unary + drops thread_id's TEXT affinity so the parent expression index applies.
Expand Down
2 changes: 2 additions & 0 deletions apps/server/src/persistence/Migrations.ts
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ import ForkMigration019 from "./Migrations/fork/019_NetworkAppReplacement.ts";
import ForkMigration020 from "./Migrations/fork/020_RepairStringifiedModelSelectionOptions.ts";
import ForkMigration021 from "./Migrations/fork/021_UpstreamScheduledTaskProfile.ts";
import ForkMigration022 from "./Migrations/fork/022_NetworkContinuity.ts";
import ForkMigration023 from "./Migrations/fork/023_ScheduledTaskCleanup.ts";

const migrationEntries = [
[1, "OrchestrationEvents", Migration0001],
Expand Down Expand Up @@ -221,6 +222,7 @@ const forkMigrationEntries = [
[20, "RepairStringifiedModelSelectionOptions", ForkMigration020, 81],
[21, "UpstreamScheduledTaskProfile", ForkMigration021, 81],
[22, "NetworkContinuity", ForkMigration022, 81],
[23, "ScheduledTaskCleanup", ForkMigration023, 81],
] as const;

export const forkMigrationManifest = forkMigrationEntries.map(
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
import { assert, it } from "@effect/vitest";
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
import * as Effect from "effect/Effect";
import * as SqlClient from "effect/sql/SqlClient";

import { runMigrations } from "../../Migrations.ts";

const row = (id: string, overrides: Record<string, unknown>) => ({
task_id: id,
title: id,
prompt: "Run",
enabled: 1,
schedule_json: '{"type":"interval","everyMs":60000}',
project_id: "project-1",
thread_id: null,
profile_id: null,
workspace_strategy_json: '{"type":"root"}',
model_selection_json: '{"instanceId":"codex","model":"gpt-5"}',
runtime_mode: "full-access",
interaction_mode: "default",
created_by: "user",
creation_source: "web",
created_at: "2026-11-01T00:00:00.000Z",
updated_at: "2026-11-01T00:00:00.000Z",
next_run_at: null,
last_run_at: null,
last_run_status: "never",
last_run_error: null,
run_count: 0,
...overrides,
});

it.effect("adds the cleanup columns, continues profile tasks and keeps webhook tokens", () =>
Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
yield* runMigrations({ toForkMigrationInclusive: 22 });
yield* sql`INSERT INTO scheduled_tasks ${sql.insert(
row("agent", { profile_id: "reviewer", thread_id: "thread-1", thread_mode: "new_thread" }),
)}`;
yield* sql`INSERT INTO scheduled_tasks ${sql.insert(
row("fresh", { thread_mode: "new_thread" }),
)}`;
yield* sql`INSERT INTO scheduled_tasks ${sql.insert(
row("hook", {
schedule_json: '{"type":"webhook","signature":null}',
webhook_token: "token-unchanged",
}),
)}`;
yield* runMigrations();

const columns = (yield* sql<{ name: string }>`PRAGMA table_info(scheduled_tasks)`).map(
(column) => column.name,
);
for (const name of ["archived_at", "archive_reason", "failing_since", "consecutive_failures"]) {
assert.include(columns, name);
}
const rows = yield* sql<{
task_id: string;
thread_mode: string;
webhook_token: string | null;
archived_at: string | null;
consecutive_failures: number;
}>`SELECT task_id, thread_mode, webhook_token, archived_at, consecutive_failures
FROM scheduled_tasks ORDER BY task_id`;
assert.deepEqual(
rows.map((entry) => [entry.task_id, entry.thread_mode, entry.webhook_token]),
[
["agent", "continue", null],
["fresh", "new_thread", null],
["hook", "new_thread", "token-unchanged"],
],
);
assert.isTrue(rows.every((entry) => entry.archived_at === null));
assert.isTrue(rows.every((entry) => entry.consecutive_failures === 0));
}).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
);
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
import * as Effect from "effect/Effect";
import * as SqlClient from "effect/sql/SqlClient";

/** Columns the stale-task sweep reads and writes, plus the mode legacy profile imports missed. */
export default Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN archived_at TEXT`;
yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN archive_reason TEXT`;
yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN failing_since TEXT`;
yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN consecutive_failures INTEGER NOT NULL DEFAULT 0`;
// An agent task always reuses the thread its first run launched.
yield* sql`UPDATE scheduled_tasks SET thread_mode = 'continue' WHERE profile_id IS NOT NULL AND thread_mode = 'new_thread'`;
});
2 changes: 2 additions & 0 deletions apps/server/src/scheduledTasks/LegacyScheduledTasks.ts
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,8 @@ export const importLegacyScheduledTasks = Effect.fn("importLegacyScheduledTasks"
project_id: row.project_id,
thread_id: row.thread_id,
profile_id: row.profile_id,
// Every fork V1 task runs as an agent, which reuses the thread its first run launched.
thread_mode: "continue",
workspace_strategy_json: encodeWorkspaceStrategy({ type: "root" }),
model_selection_json: encodeModelSelection(profile?.modelSelection ?? fallbackModel),
runtime_mode: runtimeMode,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,7 @@ it.effect("imports fork V1 scheduled tasks into the V2 table and drops the legac
});
assert.equal(task?.profileId, "reviewer");
assert.equal(task?.threadId, "thread-v1");
assert.equal(task?.threadMode, "continue");
assert.equal(task?.lastRunStatus, "succeeded");
assert.equal(task?.runCount, 4);
assert.equal(task?.nextRunAt, "2026-09-26T00:00:00.000Z");
Expand Down
Loading
Loading