fix(scheduled-tasks): keep task fields a server build does not know
Every project-config write re-serialized normalized tasks, so a server that shares the config file but predates a field (goal, auto-accept) stripped it the first time any task ran. Untouched tasks now go back to disk verbatim, a state update swaps only `state`, and only a deliberately replaced task is serialized from the normalized shape.
This commit is contained in:
@@ -499,11 +499,20 @@ export const createProjectConfigRuntime = (deps) => {
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// Normalized tasks for reading, plus the raw on-disk record of each one for
|
||||||
|
// writing back. Normalization only keeps the fields THIS build knows, so a
|
||||||
|
// write that re-serialized normalized tasks would strip every field added
|
||||||
|
// by a newer build (or a newer UI) the moment an older server touched the
|
||||||
|
// file — a goal or auto-accept setting silently lost after a task ran.
|
||||||
|
// Writers therefore persist untouched tasks from `rawTasksByID` verbatim and
|
||||||
|
// only serialize a normalized task where the task itself was deliberately
|
||||||
|
// replaced.
|
||||||
const readProjectConfigFromDisk = async (projectID) => {
|
const readProjectConfigFromDisk = async (projectID) => {
|
||||||
const parsed = await readRawProjectConfigFromDisk(projectID);
|
const parsed = await readRawProjectConfigFromDisk(projectID);
|
||||||
const tasksRaw = Array.isArray(parsed.scheduledTasks) ? parsed.scheduledTasks : [];
|
const tasksRaw = Array.isArray(parsed.scheduledTasks) ? parsed.scheduledTasks : [];
|
||||||
const now = Date.now();
|
const now = Date.now();
|
||||||
const scheduledTasks = [];
|
const scheduledTasks = [];
|
||||||
|
const rawTasksByID = new Map();
|
||||||
for (const task of tasksRaw) {
|
for (const task of tasksRaw) {
|
||||||
try {
|
try {
|
||||||
const normalized = normalizeTaskForStorage(task, {
|
const normalized = normalizeTaskForStorage(task, {
|
||||||
@@ -514,15 +523,31 @@ export const createProjectConfigRuntime = (deps) => {
|
|||||||
refreshUpdatedAt: false,
|
refreshUpdatedAt: false,
|
||||||
});
|
});
|
||||||
scheduledTasks.push(normalized);
|
scheduledTasks.push(normalized);
|
||||||
|
rawTasksByID.set(normalized.id, task);
|
||||||
} catch {
|
} catch {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return {
|
return {
|
||||||
version: PROJECT_CONFIG_VERSION,
|
version: PROJECT_CONFIG_VERSION,
|
||||||
scheduledTasks,
|
scheduledTasks,
|
||||||
|
rawTasksByID,
|
||||||
};
|
};
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// The list to write: tasks this write replaced go out normalized; every
|
||||||
|
// other task goes out exactly as stored, fields unknown to this build
|
||||||
|
// included. A state-only update counts as untouched — only its `state` is
|
||||||
|
// swapped onto the stored record. Callers keep working with (and returning)
|
||||||
|
// the normalized tasks; only the bytes on disk differ.
|
||||||
|
const toStoredTasks = (config, tasks, { replacedIDs = new Set(), stateUpdatedID = null } = {}) => (
|
||||||
|
tasks.map((task) => {
|
||||||
|
if (replacedIDs.has(task.id)) return task;
|
||||||
|
const stored = config.rawTasksByID.get(task.id);
|
||||||
|
if (!stored) return task;
|
||||||
|
return task.id === stateUpdatedID ? { ...stored, state: task.state } : stored;
|
||||||
|
})
|
||||||
|
);
|
||||||
|
|
||||||
const writeProjectConfigToDisk = async (projectID, config) => {
|
const writeProjectConfigToDisk = async (projectID, config) => {
|
||||||
const filePath = resolveProjectConfigPath(projectID);
|
const filePath = resolveProjectConfigPath(projectID);
|
||||||
const parentDirectory = path.dirname(filePath);
|
const parentDirectory = path.dirname(filePath);
|
||||||
@@ -607,7 +632,7 @@ export const createProjectConfigRuntime = (deps) => {
|
|||||||
|
|
||||||
const nextConfig = {
|
const nextConfig = {
|
||||||
version: PROJECT_CONFIG_VERSION,
|
version: PROJECT_CONFIG_VERSION,
|
||||||
scheduledTasks: nextTasks,
|
scheduledTasks: toStoredTasks(current, nextTasks, { replacedIDs: new Set([normalizedTask.id]) }),
|
||||||
};
|
};
|
||||||
await writeProjectConfigToDisk(projectID, nextConfig);
|
await writeProjectConfigToDisk(projectID, nextConfig);
|
||||||
|
|
||||||
@@ -633,7 +658,7 @@ export const createProjectConfigRuntime = (deps) => {
|
|||||||
if (deleted) {
|
if (deleted) {
|
||||||
await writeProjectConfigToDisk(projectID, {
|
await writeProjectConfigToDisk(projectID, {
|
||||||
version: PROJECT_CONFIG_VERSION,
|
version: PROJECT_CONFIG_VERSION,
|
||||||
scheduledTasks: nextTasks,
|
scheduledTasks: toStoredTasks(current, nextTasks),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -676,7 +701,7 @@ export const createProjectConfigRuntime = (deps) => {
|
|||||||
|
|
||||||
await writeProjectConfigToDisk(projectID, {
|
await writeProjectConfigToDisk(projectID, {
|
||||||
version: PROJECT_CONFIG_VERSION,
|
version: PROJECT_CONFIG_VERSION,
|
||||||
scheduledTasks: nextTasks,
|
scheduledTasks: toStoredTasks(current, nextTasks, { stateUpdatedID: nextTask.id }),
|
||||||
});
|
});
|
||||||
|
|
||||||
return {
|
return {
|
||||||
@@ -737,7 +762,7 @@ export const createProjectConfigRuntime = (deps) => {
|
|||||||
|
|
||||||
await writeProjectConfigToDisk(projectID, {
|
await writeProjectConfigToDisk(projectID, {
|
||||||
version: PROJECT_CONFIG_VERSION,
|
version: PROJECT_CONFIG_VERSION,
|
||||||
scheduledTasks: nextTasks,
|
scheduledTasks: toStoredTasks(current, nextTasks, { stateUpdatedID: nextTask.id }),
|
||||||
});
|
});
|
||||||
|
|
||||||
return {
|
return {
|
||||||
@@ -797,6 +822,7 @@ export const createProjectConfigRuntime = (deps) => {
|
|||||||
|
|
||||||
const consumedLoopPaths = new Set();
|
const consumedLoopPaths = new Set();
|
||||||
const nextTasks = [];
|
const nextTasks = [];
|
||||||
|
const replacedIDs = new Set();
|
||||||
for (const task of tasks) {
|
for (const task of tasks) {
|
||||||
if (task.loopFile && !activeLoopFilePaths.has(task.loopFile)) {
|
if (task.loopFile && !activeLoopFilePaths.has(task.loopFile)) {
|
||||||
// The driving loop file was removed (or renamed) — unschedule.
|
// The driving loop file was removed (or renamed) — unschedule.
|
||||||
@@ -828,6 +854,7 @@ export const createProjectConfigRuntime = (deps) => {
|
|||||||
},
|
},
|
||||||
);
|
);
|
||||||
nextTasks.push(adopted);
|
nextTasks.push(adopted);
|
||||||
|
replacedIDs.add(adopted.id);
|
||||||
pendingLoops.delete(loop.definition.name);
|
pendingLoops.delete(loop.definition.name);
|
||||||
if (task.loopFile) {
|
if (task.loopFile) {
|
||||||
consumedLoopPaths.add(task.loopFile);
|
consumedLoopPaths.add(task.loopFile);
|
||||||
@@ -863,6 +890,7 @@ export const createProjectConfigRuntime = (deps) => {
|
|||||||
},
|
},
|
||||||
);
|
);
|
||||||
nextTasks.push(created);
|
nextTasks.push(created);
|
||||||
|
replacedIDs.add(created.id);
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
console.warn(`[scheduled-tasks] skipped loop ${loop.filePath}:`, error?.message ?? error);
|
console.warn(`[scheduled-tasks] skipped loop ${loop.filePath}:`, error?.message ?? error);
|
||||||
}
|
}
|
||||||
@@ -870,7 +898,7 @@ export const createProjectConfigRuntime = (deps) => {
|
|||||||
|
|
||||||
await writeProjectConfigToDisk(projectID, {
|
await writeProjectConfigToDisk(projectID, {
|
||||||
version: PROJECT_CONFIG_VERSION,
|
version: PROJECT_CONFIG_VERSION,
|
||||||
scheduledTasks: nextTasks,
|
scheduledTasks: toStoredTasks(current, nextTasks, { replacedIDs }),
|
||||||
});
|
});
|
||||||
|
|
||||||
return nextTasks;
|
return nextTasks;
|
||||||
|
|||||||
@@ -14,6 +14,7 @@ const createRuntime = async () => {
|
|||||||
});
|
});
|
||||||
return {
|
return {
|
||||||
runtime,
|
runtime,
|
||||||
|
tempRoot,
|
||||||
cleanup: async () => {
|
cleanup: async () => {
|
||||||
await rm(tempRoot, { recursive: true, force: true });
|
await rm(tempRoot, { recursive: true, force: true });
|
||||||
},
|
},
|
||||||
@@ -472,6 +473,90 @@ describe('project-config loop reconciliation', () => {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe('fields this build does not know', () => {
|
||||||
|
// Simulates a config written by a newer build (or a newer UI): the task
|
||||||
|
// carries execution and state fields normalization here has never heard of.
|
||||||
|
const seedForeignTask = async (runtime, tempRoot) => {
|
||||||
|
const created = await runtime.upsertScheduledTask('project-test', {
|
||||||
|
name: 'Nightly digest',
|
||||||
|
enabled: true,
|
||||||
|
schedule: { kind: 'daily', time: '09:30', timezone: 'UTC' },
|
||||||
|
execution: { prompt: 'Summarize', providerID: 'openai', modelID: 'gpt-4.1', goalEnabled: true },
|
||||||
|
});
|
||||||
|
const filePath = path.join(tempRoot, 'project-test.json');
|
||||||
|
const stored = JSON.parse(await readFile(filePath, 'utf8'));
|
||||||
|
stored.scheduledTasks[0].execution.futureExecutionField = 'keep me';
|
||||||
|
stored.scheduledTasks[0].state.futureStateField = 42;
|
||||||
|
stored.scheduledTasks[0].futureTopLevelField = true;
|
||||||
|
await writeFile(filePath, JSON.stringify(stored, null, 2), 'utf8');
|
||||||
|
return { id: created.task.id, filePath };
|
||||||
|
};
|
||||||
|
|
||||||
|
const readStoredTask = async (filePath, id) => {
|
||||||
|
const stored = JSON.parse(await readFile(filePath, 'utf8'));
|
||||||
|
return stored.scheduledTasks.find((task) => task.id === id);
|
||||||
|
};
|
||||||
|
|
||||||
|
it('survive a state update after a run, and the claim update', async () => {
|
||||||
|
const { runtime, tempRoot, cleanup } = await createRuntime();
|
||||||
|
try {
|
||||||
|
const { id, filePath } = await seedForeignTask(runtime, tempRoot);
|
||||||
|
|
||||||
|
await runtime.updateScheduledTaskState('project-test', id, { lastStatus: 'success', lastRunAt: 1000 });
|
||||||
|
let stored = await readStoredTask(filePath, id);
|
||||||
|
expect(stored.execution.futureExecutionField).toBe('keep me');
|
||||||
|
expect(stored.execution.goalEnabled).toBe(true);
|
||||||
|
expect(stored.futureTopLevelField).toBe(true);
|
||||||
|
expect(stored.state.lastStatus).toBe('success');
|
||||||
|
|
||||||
|
await runtime.updateScheduledTaskStateIf('project-test', id, () => true, { lastScheduledFor: 5000 });
|
||||||
|
stored = await readStoredTask(filePath, id);
|
||||||
|
expect(stored.execution.futureExecutionField).toBe('keep me');
|
||||||
|
expect(stored.state.lastScheduledFor).toBe(5000);
|
||||||
|
} finally {
|
||||||
|
await cleanup();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it('survive writes that replace or delete a different task, and a loop sync', async () => {
|
||||||
|
const { runtime, tempRoot, cleanup } = await createRuntime();
|
||||||
|
try {
|
||||||
|
const { id, filePath } = await seedForeignTask(runtime, tempRoot);
|
||||||
|
|
||||||
|
const other = await runtime.upsertScheduledTask('project-test', {
|
||||||
|
id: 'other-task',
|
||||||
|
name: 'Other',
|
||||||
|
enabled: true,
|
||||||
|
schedule: { kind: 'daily', time: '10:00', timezone: 'UTC' },
|
||||||
|
execution: { prompt: 'Other', providerID: 'openai', modelID: 'gpt-4.1' },
|
||||||
|
});
|
||||||
|
expect((await readStoredTask(filePath, id)).execution.futureExecutionField).toBe('keep me');
|
||||||
|
|
||||||
|
await runtime.deleteScheduledTask('project-test', other.task.id);
|
||||||
|
expect((await readStoredTask(filePath, id)).execution.futureExecutionField).toBe('keep me');
|
||||||
|
|
||||||
|
await runtime.reconcileLoopTasks('project-test', []);
|
||||||
|
expect((await readStoredTask(filePath, id)).execution.futureExecutionField).toBe('keep me');
|
||||||
|
} finally {
|
||||||
|
await cleanup();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it('are dropped only when the task itself is deliberately saved', async () => {
|
||||||
|
const { runtime, tempRoot, cleanup } = await createRuntime();
|
||||||
|
try {
|
||||||
|
const { id, filePath } = await seedForeignTask(runtime, tempRoot);
|
||||||
|
const [task] = await runtime.listScheduledTasks('project-test');
|
||||||
|
await runtime.upsertScheduledTask('project-test', { ...task, name: 'Renamed' });
|
||||||
|
const stored = await readStoredTask(filePath, id);
|
||||||
|
expect(stored.name).toBe('Renamed');
|
||||||
|
expect(stored.execution.futureExecutionField).toBeUndefined();
|
||||||
|
} finally {
|
||||||
|
await cleanup();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
it('conditionally updates state only when the predicate passes (occurrence claim)', async () => {
|
it('conditionally updates state only when the predicate passes (occurrence claim)', async () => {
|
||||||
const { runtime, cleanup } = await createRuntime();
|
const { runtime, cleanup } = await createRuntime();
|
||||||
try {
|
try {
|
||||||
|
|||||||
@@ -25,6 +25,13 @@ in shared project config under the project write lock:
|
|||||||
from the winner's persisted `nextRunAt`.
|
from the winner's persisted `nextRunAt`.
|
||||||
- Project config writes also take a cross-process `.json.lock` file so the
|
- Project config writes also take a cross-process `.json.lock` file so the
|
||||||
read-modify-write is serialized across processes, not only within one process.
|
read-modify-write is serialized across processes, not only within one process.
|
||||||
|
- The sharing processes may run different OpenChamber versions. Normalization
|
||||||
|
keeps only the fields a build knows, so every writer persists tasks it did
|
||||||
|
not change verbatim from disk and swaps only `state` onto a task whose state
|
||||||
|
it updated; a task goes out normalized only when it was deliberately
|
||||||
|
replaced (upsert, loop adoption). An older server touching the file after a
|
||||||
|
run therefore cannot strip fields a newer build added, such as a task's goal
|
||||||
|
or auto-accept settings.
|
||||||
- Lock timeout / filesystem errors on claim, manual-start, or completion state
|
- Lock timeout / filesystem errors on claim, manual-start, or completion state
|
||||||
writes always release the in-process running slot (via `finally`) and best-effort
|
writes always release the in-process running slot (via `finally`) and best-effort
|
||||||
re-arm the **next future** occurrence; they must not leave the task permanently
|
re-arm the **next future** occurrence; they must not leave the task permanently
|
||||||
|
|||||||
Reference in New Issue
Block a user