mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 12:04:41 +01:00
Ensure worktree execution starts only after activation (#9374)
## Thinking Path > - Paperclip is the open-source control plane people use to manage AI agents and their work. > - Its scheduler, routines, and heartbeat services decide when agents automatically begin work. > - Experimental per-worktree execution is useful for isolated development, but enabling it previously allowed automatic services to consider an existing backlog. > - A worktree activation must therefore create a durable eligibility boundary rather than merely toggle execution on. > - This pull request records an activation cutoff and applies it consistently to automatic routine and heartbeat dispatch. > - The result is that an enabled worktree executes only work created after its own activation, while non-worktree behavior remains unchanged. ## Linked Issues or Issue Description **Problem type:** Bug / safety regression **Summary:** Enabling experimental run execution in an existing worktree could start automatic scheduler, routine, watchdog, and heartbeat activity for work created before that worktree was explicitly armed. **Expected behavior:** A worktree that has execution enabled only considers automatically dispatched work created on or after its activation timestamp. Ambiguous activation state fails closed. Non-worktree instances keep their existing behavior. **Related public work:** Refs #8275 (runtime worktree policy gating); this PR adds an activation-time boundary for automatic execution rather than changing the general runtime policy. ## What Changed - Persist a worktree execution activation timestamp and originating instance ID; stamp them only when the experimental toggle changes from disabled to enabled. - Resolve activation state fail-closed when the cutoff is missing, invalid, disabled, or belongs to another instance. - Gate automatic routine scheduling, webhooks, watchdog activity, and heartbeat selection at the activation cutoff; manual runs remain available. - Share the canonical worktree truthy-environment helper across routine dispatch and agent inbox filtering. - Add cutoff and truthy-runtime regression coverage, plus experimental-settings UI states that explain armed and suppressed execution. ## Verification - `pnpm exec vitest run server/src/__tests__/routines-service.test.ts server/src/__tests__/instance-settings-service.test.ts` — passes: 2 files, 60 tests. - `pnpm --filter @paperclipai/server typecheck` — passes. - Existing CI completed successfully before the follow-up review fixes; this branch was rebased onto the latest `origin/master` before retesting. ## Risks - **Behavioral:** Automatic worktree execution is intentionally more restrictive; pre-existing work is suppressed until newly created after activation. - **Operational:** A malformed or cross-instance activation record fails closed, requiring an operator to disable and re-enable the experimental toggle on the intended worktree. - **Compatibility:** The worktree environment now accepts all canonical truthy values (`1`, `true`, `yes`, and `on`) consistently; non-worktree instances are unaffected. - **Branch metadata:** This existing execution-workspace branch predates the current naming rule and cannot be renamed under this task's workspace contract; the code and PR title do not include internal ticket references. > `ROADMAP.md` was checked; this targeted execution-safety fix does not duplicate planned core work. ## Model Used - Anthropic Claude Code — assisted with the original implementation; exact model identifier and context window were not recorded in the repository metadata. - OpenAI Codex CLI — assisted with PR preparation and review fixes; exact model identifier and context window are not exposed in this execution environment. Used with terminal tooling, code editing, and targeted test execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [ ] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
22 files changed
+1082
-69
No files matched your search
@@ -161,7 +161,10 @@ import {
|
||||
WORKSPACE_WORKTREE_REQUIRES_PROJECT_MESSAGE,
|
||||
WORKSPACE_WORKTREE_REQUIRES_PROJECT_REMEDIATION,
|
||||
} from "./execution-workspace-policy.js";
|
||||
import { instanceSettingsService } from "./instance-settings.js";
|
||||
import {
|
||||
instanceSettingsService,
|
||||
resolveWorktreeRunExecutionActivation,
|
||||
} from "./instance-settings.js";
|
||||
import {
|
||||
evaluateExecutionAllowlist,
|
||||
isExecutionForcedToKubernetes,
|
||||
@@ -5018,30 +5021,45 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
// Only worktree runtimes ever read the setting; a short TTL keeps the hot-path
|
||||
// suppression checks off the DB, and a read failure falls back to prior/default
|
||||
// (fail closed to suppression).
|
||||
let cachedWorktreeRunExecutionOverride: { value: boolean; at: number } = { value: false, at: 0 };
|
||||
let cachedWorktreeRunExecutionOverride: { allowed: boolean; cutoff: Date | null; at: number } = {
|
||||
allowed: false,
|
||||
cutoff: null,
|
||||
at: 0,
|
||||
};
|
||||
const WORKTREE_RUN_EXECUTION_OVERRIDE_TTL_MS = 3_000;
|
||||
const resolveWorktreeRunExecutionOverride = async (): Promise<boolean> => {
|
||||
if (!inWorktreeRuntime) return false;
|
||||
const resolveWorktreeRunExecutionOverride = async () => {
|
||||
if (!inWorktreeRuntime) return { allowed: false, cutoff: null };
|
||||
const now = Date.now();
|
||||
if (now - cachedWorktreeRunExecutionOverride.at < WORKTREE_RUN_EXECUTION_OVERRIDE_TTL_MS) {
|
||||
return cachedWorktreeRunExecutionOverride.value;
|
||||
return cachedWorktreeRunExecutionOverride;
|
||||
}
|
||||
try {
|
||||
const experimental = await instanceSettings.getExperimental();
|
||||
const activation = resolveWorktreeRunExecutionActivation(
|
||||
await instanceSettings.getExperimental(),
|
||||
runtimeEnv.PAPERCLIP_INSTANCE_ID?.trim() || null,
|
||||
);
|
||||
const cutoff = activation.armed ? new Date(activation.cutoff) : null;
|
||||
cachedWorktreeRunExecutionOverride = {
|
||||
value: experimental.enableWorktreeRunExecution === true,
|
||||
allowed: Boolean(activation.armed && cutoff && !Number.isNaN(cutoff.getTime())),
|
||||
cutoff: cutoff && !Number.isNaN(cutoff.getTime()) ? cutoff : null,
|
||||
at: now,
|
||||
};
|
||||
} catch {
|
||||
// Keep the prior (default-false) value so a settings read failure fails
|
||||
// closed to the safe suppressed state.
|
||||
}
|
||||
return cachedWorktreeRunExecutionOverride.value;
|
||||
return cachedWorktreeRunExecutionOverride;
|
||||
};
|
||||
const getSchedulingSuppression = async () =>
|
||||
resolveHeartbeatSchedulingSuppression(runtimeEnv, {
|
||||
allowWorktreeRunExecution: await resolveWorktreeRunExecutionOverride(),
|
||||
const getSchedulingSuppression = async () => {
|
||||
const override = await resolveWorktreeRunExecutionOverride();
|
||||
return resolveHeartbeatSchedulingSuppression(runtimeEnv, {
|
||||
allowWorktreeRunExecution: override.allowed,
|
||||
});
|
||||
};
|
||||
const getWorktreeExecutionCutoff = async () => {
|
||||
const override = await resolveWorktreeRunExecutionOverride();
|
||||
return override.allowed ? override.cutoff : null;
|
||||
};
|
||||
|
||||
const runLogStore = getRunLogStore();
|
||||
const secretsSvc = secretService(db);
|
||||
@@ -9462,6 +9480,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
}
|
||||
|
||||
async function promoteDueScheduledRetries(now = new Date()) {
|
||||
const cutoff = await getWorktreeExecutionCutoff();
|
||||
const dueRuns = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
@@ -9469,6 +9488,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
and(
|
||||
eq(heartbeatRuns.status, "scheduled_retry"),
|
||||
lte(heartbeatRuns.scheduledRetryAt, now),
|
||||
cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined,
|
||||
),
|
||||
)
|
||||
.orderBy(asc(heartbeatRuns.scheduledRetryAt), asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id))
|
||||
@@ -10745,6 +10765,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
|
||||
async function resumeQueuedRuns() {
|
||||
if ((await getSchedulingSuppression()).suppressed) return;
|
||||
const cutoff = await getWorktreeExecutionCutoff();
|
||||
|
||||
const queuedRuns = await db
|
||||
.select({ agentId: heartbeatRuns.agentId })
|
||||
@@ -10753,6 +10774,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
.where(and(
|
||||
eq(heartbeatRuns.status, "queued"),
|
||||
eq(companies.status, "active"),
|
||||
cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined,
|
||||
));
|
||||
|
||||
const agentIds = [...new Set(queuedRuns.map((r) => r.agentId))];
|
||||
@@ -10762,7 +10784,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
}
|
||||
|
||||
async function reconcileStrandedAssignedIssues() {
|
||||
return recovery.reconcileStrandedAssignedIssues();
|
||||
return recovery.reconcileStrandedAssignedIssues({ issueCreatedAtGte: await getWorktreeExecutionCutoff() });
|
||||
}
|
||||
|
||||
async function sweepStaleIssueLocks() {
|
||||
@@ -10783,15 +10805,15 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
}
|
||||
|
||||
async function scanSilentActiveRuns(opts?: { now?: Date; companyId?: string }) {
|
||||
return recovery.scanSilentActiveRuns(opts);
|
||||
return recovery.scanSilentActiveRuns({ ...opts, issueCreatedAtGte: await getWorktreeExecutionCutoff() });
|
||||
}
|
||||
|
||||
async function reconcileProductivityReviews(opts?: { now?: Date; companyId?: string }) {
|
||||
return productivityReviews.reconcileProductivityReviews(opts);
|
||||
return productivityReviews.reconcileProductivityReviews({ ...opts, issueCreatedAtGte: await getWorktreeExecutionCutoff() });
|
||||
}
|
||||
|
||||
async function reconcileTaskWatchdogs(opts?: { companyId?: string | null; runId?: string | null }) {
|
||||
return taskWatchdogs.reconcileTaskWatchdogs(opts);
|
||||
return taskWatchdogs.reconcileTaskWatchdogs({ ...opts, issueCreatedAtGte: await getWorktreeExecutionCutoff() });
|
||||
}
|
||||
|
||||
async function buildRunOutputSilence(
|
||||
@@ -10813,7 +10835,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
force?: boolean;
|
||||
lookbackHours?: number;
|
||||
}) {
|
||||
return recovery.reconcileIssueGraphLiveness(opts);
|
||||
return recovery.reconcileIssueGraphLiveness({ ...opts, issueCreatedAtGte: await getWorktreeExecutionCutoff() });
|
||||
}
|
||||
|
||||
async function updateRuntimeState(
|
||||
@@ -10873,6 +10895,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
|
||||
async function startNextQueuedRunForAgent(agentId: string) {
|
||||
if ((await getSchedulingSuppression()).suppressed) return [];
|
||||
const cutoff = await getWorktreeExecutionCutoff();
|
||||
|
||||
return withAgentStartLock(agentId, async () => {
|
||||
const agent = await getAgent(agentId);
|
||||
@@ -10892,7 +10915,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
const queuedRuns = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.agentId, agentId), eq(heartbeatRuns.status, "queued")))
|
||||
.where(and(
|
||||
eq(heartbeatRuns.agentId, agentId),
|
||||
eq(heartbeatRuns.status, "queued"),
|
||||
cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined,
|
||||
))
|
||||
.orderBy(asc(heartbeatRuns.createdAt));
|
||||
if (queuedRuns.length === 0) return [];
|
||||
|
||||
@@ -14304,6 +14331,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
return null;
|
||||
}
|
||||
|
||||
const worktreeExecutionCutoff = opts.requestedByActorType === "user"
|
||||
? null
|
||||
: await getWorktreeExecutionCutoff();
|
||||
|
||||
const company = await db
|
||||
.select({ status: companies.status })
|
||||
.from(companies)
|
||||
@@ -14369,11 +14400,19 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
? or(eq(issues.id, issueId), eq(issues.identifier, issueId.toUpperCase()))
|
||||
: eq(issues.identifier, issueId.toUpperCase());
|
||||
const resolvedIssue = await db
|
||||
.select({ id: issues.id, projectId: issues.projectId })
|
||||
.select({ id: issues.id, projectId: issues.projectId, createdAt: issues.createdAt })
|
||||
.from(issues)
|
||||
.where(and(eq(issues.companyId, agent.companyId), idMatch))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (resolvedIssue) {
|
||||
if (worktreeExecutionCutoff && resolvedIssue.createdAt < worktreeExecutionCutoff) {
|
||||
await writeSkippedHeartbeatRequest("heartbeat.worktree_execution_cutoff", {
|
||||
reason: "worktree_execution_cutoff",
|
||||
cutoff: worktreeExecutionCutoff.toISOString(),
|
||||
issueId: resolvedIssue.id,
|
||||
});
|
||||
return null;
|
||||
}
|
||||
projectId = resolvedIssue.projectId ?? null;
|
||||
// Canonicalize context to the UUID so downstream lookups always use UUID
|
||||
if (resolvedIssue.id !== issueId) {
|
||||
@@ -14548,6 +14587,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
executionRunId: issues.executionRunId,
|
||||
executionAgentNameKey: issues.executionAgentNameKey,
|
||||
createdAt: issues.createdAt,
|
||||
})
|
||||
.from(issues)
|
||||
.where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId)))
|
||||
@@ -14570,6 +14610,30 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
return { kind: "skipped" as const };
|
||||
}
|
||||
|
||||
if (worktreeExecutionCutoff && issue.createdAt < worktreeExecutionCutoff) {
|
||||
await tx.insert(agentWakeupRequests).values({
|
||||
companyId: agent.companyId,
|
||||
agentId,
|
||||
source,
|
||||
triggerDetail,
|
||||
reason: "heartbeat.worktree_execution_cutoff",
|
||||
payload: {
|
||||
...(payload ?? {}),
|
||||
heartbeatSkip: {
|
||||
reason: "worktree_execution_cutoff",
|
||||
cutoff: worktreeExecutionCutoff.toISOString(),
|
||||
issueId: issue.id,
|
||||
},
|
||||
},
|
||||
status: "skipped",
|
||||
requestedByActorType: opts.requestedByActorType ?? null,
|
||||
requestedByActorId: opts.requestedByActorId ?? null,
|
||||
idempotencyKey: opts.idempotencyKey ?? null,
|
||||
finishedAt: new Date(),
|
||||
});
|
||||
return { kind: "skipped" as const };
|
||||
}
|
||||
|
||||
const cancelStaleScheduledRetry = async (scheduledRun: typeof heartbeatRuns.$inferSelect) => {
|
||||
const issueCancelled = issue.status === "cancelled";
|
||||
if (
|
||||
@@ -15952,6 +16016,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
skipped: 0,
|
||||
};
|
||||
}
|
||||
const cutoff = await getWorktreeExecutionCutoff();
|
||||
|
||||
const allAgents = await db
|
||||
.select({ ...getTableColumns(agents) })
|
||||
@@ -15969,6 +16034,21 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
const policy = parseHeartbeatPolicy(agent);
|
||||
if (!policy.enabled || policy.intervalSec <= 0) continue;
|
||||
|
||||
if (cutoff) {
|
||||
const eligibleIssue = await db
|
||||
.select({ id: issues.id })
|
||||
.from(issues)
|
||||
.where(and(
|
||||
eq(issues.companyId, agent.companyId),
|
||||
eq(issues.assigneeAgentId, agent.id),
|
||||
inArray(issues.status, ["todo", "in_progress"]),
|
||||
gte(issues.createdAt, cutoff),
|
||||
))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!eligibleIssue) continue;
|
||||
}
|
||||
|
||||
checked += 1;
|
||||
const baseline = new Date(agent.lastHeartbeatAt ?? agent.createdAt).getTime();
|
||||
const elapsedMs = now.getTime() - baseline;
|
||||
|
||||
@@ -18,6 +18,164 @@ import { eq } from "drizzle-orm";
|
||||
const DEFAULT_SINGLETON_KEY = "default";
|
||||
const instanceGeneralSettingsStorageSchema = instanceGeneralSettingsSchema.strip();
|
||||
const instanceExperimentalSettingsStorageSchema = instanceExperimentalSettingsSchema.strip();
|
||||
const TRUTHY_RUNTIME_ENV_VALUES = new Set(["1", "true", "yes", "on"]);
|
||||
|
||||
interface InstanceSettingsServiceOptions {
|
||||
runtimeEnv?: Record<string, string | undefined>;
|
||||
now?: () => Date;
|
||||
}
|
||||
|
||||
type WorktreeRunExecutionSuppressedReason =
|
||||
| "not_worktree_runtime"
|
||||
| "flag_disabled"
|
||||
| "missing_cutoff"
|
||||
| "missing_instance_id"
|
||||
| "instance_id_mismatch"
|
||||
| "settings_read_error";
|
||||
|
||||
export type WorktreeRunExecutionActivationState =
|
||||
| {
|
||||
armed: true;
|
||||
cutoff: string;
|
||||
activationInstanceId: string;
|
||||
reason: null;
|
||||
}
|
||||
| {
|
||||
armed: false;
|
||||
cutoff: null;
|
||||
activationInstanceId: string | null;
|
||||
reason: WorktreeRunExecutionSuppressedReason;
|
||||
};
|
||||
|
||||
export function isTruthyRuntimeEnvValue(value: string | undefined) {
|
||||
return typeof value === "string" && TRUTHY_RUNTIME_ENV_VALUES.has(value.trim().toLowerCase());
|
||||
}
|
||||
|
||||
function getRuntimeInstanceId(env: Record<string, string | undefined>) {
|
||||
const instanceId = env.PAPERCLIP_INSTANCE_ID?.trim();
|
||||
return instanceId ? instanceId : null;
|
||||
}
|
||||
|
||||
function stripServerManagedExperimentalPatchFields(
|
||||
patch: PatchInstanceExperimentalSettings | Record<string, unknown>,
|
||||
): PatchInstanceExperimentalSettings {
|
||||
const {
|
||||
worktreeRunExecutionActivatedAt: _ignoredActivatedAt,
|
||||
worktreeRunExecutionActivationInstanceId: _ignoredActivationInstanceId,
|
||||
...patchable
|
||||
} = patch as Record<string, unknown>;
|
||||
return patchable as PatchInstanceExperimentalSettings;
|
||||
}
|
||||
|
||||
export function applyExperimentalSettingsPatch(
|
||||
current: unknown,
|
||||
patch: PatchInstanceExperimentalSettings | Record<string, unknown>,
|
||||
options: InstanceSettingsServiceOptions = {},
|
||||
): InstanceExperimentalSettings {
|
||||
const previousExperimental = normalizeExperimentalSettings(current);
|
||||
const patchable = stripServerManagedExperimentalPatchFields(patch);
|
||||
const nextExperimental = normalizeExperimentalSettings({
|
||||
...previousExperimental,
|
||||
...patchable,
|
||||
});
|
||||
const hasWorktreeRunExecutionPatch = Object.prototype.hasOwnProperty.call(
|
||||
patchable,
|
||||
"enableWorktreeRunExecution",
|
||||
);
|
||||
|
||||
if (!hasWorktreeRunExecutionPatch) {
|
||||
return nextExperimental;
|
||||
}
|
||||
|
||||
if (nextExperimental.enableWorktreeRunExecution !== true) {
|
||||
return {
|
||||
...nextExperimental,
|
||||
worktreeRunExecutionActivatedAt: null,
|
||||
worktreeRunExecutionActivationInstanceId: null,
|
||||
};
|
||||
}
|
||||
|
||||
if (previousExperimental.enableWorktreeRunExecution === true) {
|
||||
return nextExperimental;
|
||||
}
|
||||
|
||||
const runtimeEnv = options.runtimeEnv ?? process.env;
|
||||
if (!isTruthyRuntimeEnvValue(runtimeEnv.PAPERCLIP_IN_WORKTREE)) {
|
||||
return nextExperimental;
|
||||
}
|
||||
|
||||
return {
|
||||
...nextExperimental,
|
||||
worktreeRunExecutionActivatedAt: (options.now ?? (() => new Date()))().toISOString(),
|
||||
worktreeRunExecutionActivationInstanceId: getRuntimeInstanceId(runtimeEnv),
|
||||
};
|
||||
}
|
||||
|
||||
function suppressWorktreeRunExecution(
|
||||
reason: WorktreeRunExecutionSuppressedReason,
|
||||
activationInstanceId: string | null = null,
|
||||
): WorktreeRunExecutionActivationState {
|
||||
return {
|
||||
armed: false,
|
||||
cutoff: null,
|
||||
activationInstanceId,
|
||||
reason,
|
||||
};
|
||||
}
|
||||
|
||||
export function resolveWorktreeRunExecutionActivation(
|
||||
experimental: InstanceExperimentalSettings,
|
||||
currentInstanceId: string | null | undefined,
|
||||
): WorktreeRunExecutionActivationState {
|
||||
if (experimental.enableWorktreeRunExecution !== true) {
|
||||
return suppressWorktreeRunExecution(
|
||||
"flag_disabled",
|
||||
experimental.worktreeRunExecutionActivationInstanceId,
|
||||
);
|
||||
}
|
||||
if (!experimental.worktreeRunExecutionActivatedAt) {
|
||||
return suppressWorktreeRunExecution(
|
||||
"missing_cutoff",
|
||||
experimental.worktreeRunExecutionActivationInstanceId,
|
||||
);
|
||||
}
|
||||
if (!currentInstanceId) {
|
||||
return suppressWorktreeRunExecution(
|
||||
"missing_instance_id",
|
||||
experimental.worktreeRunExecutionActivationInstanceId,
|
||||
);
|
||||
}
|
||||
if (experimental.worktreeRunExecutionActivationInstanceId !== currentInstanceId) {
|
||||
return suppressWorktreeRunExecution(
|
||||
"instance_id_mismatch",
|
||||
experimental.worktreeRunExecutionActivationInstanceId,
|
||||
);
|
||||
}
|
||||
return {
|
||||
armed: true,
|
||||
cutoff: experimental.worktreeRunExecutionActivatedAt,
|
||||
activationInstanceId: currentInstanceId,
|
||||
reason: null,
|
||||
};
|
||||
}
|
||||
|
||||
export async function resolveWorktreeRunExecutionActivationState(options: {
|
||||
getExperimental: () => Promise<InstanceExperimentalSettings>;
|
||||
runtimeEnv?: Record<string, string | undefined>;
|
||||
}): Promise<WorktreeRunExecutionActivationState> {
|
||||
const runtimeEnv = options.runtimeEnv ?? process.env;
|
||||
if (!isTruthyRuntimeEnvValue(runtimeEnv.PAPERCLIP_IN_WORKTREE)) {
|
||||
return suppressWorktreeRunExecution("not_worktree_runtime");
|
||||
}
|
||||
try {
|
||||
return resolveWorktreeRunExecutionActivation(
|
||||
await options.getExperimental(),
|
||||
getRuntimeInstanceId(runtimeEnv),
|
||||
);
|
||||
} catch {
|
||||
return suppressWorktreeRunExecution("settings_read_error");
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeGeneralSettings(raw: unknown): InstanceGeneralSettings {
|
||||
const parsed = instanceGeneralSettingsStorageSchema.safeParse(raw ?? {});
|
||||
@@ -63,6 +221,9 @@ export function normalizeExperimentalSettings(raw: unknown): InstanceExperimenta
|
||||
enableWorkspaceBranchReconcileForward: parsed.data.enableWorkspaceBranchReconcileForward ?? true,
|
||||
enableWorkspaceDirtyQuarantineRepair: parsed.data.enableWorkspaceDirtyQuarantineRepair ?? true,
|
||||
enableWorktreeRunExecution: parsed.data.enableWorktreeRunExecution ?? false,
|
||||
worktreeRunExecutionActivatedAt: parsed.data.worktreeRunExecutionActivatedAt ?? null,
|
||||
worktreeRunExecutionActivationInstanceId:
|
||||
parsed.data.worktreeRunExecutionActivationInstanceId ?? null,
|
||||
issueGraphLivenessAutoRecoveryLookbackHours:
|
||||
parsed.data.issueGraphLivenessAutoRecoveryLookbackHours ??
|
||||
DEFAULT_ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_LOOKBACK_HOURS,
|
||||
@@ -88,6 +249,8 @@ export function normalizeExperimentalSettings(raw: unknown): InstanceExperimenta
|
||||
enableWorkspaceBranchReconcileForward: true,
|
||||
enableWorkspaceDirtyQuarantineRepair: true,
|
||||
enableWorktreeRunExecution: false,
|
||||
worktreeRunExecutionActivatedAt: null,
|
||||
worktreeRunExecutionActivationInstanceId: null,
|
||||
issueGraphLivenessAutoRecoveryLookbackHours:
|
||||
DEFAULT_ISSUE_GRAPH_LIVENESS_AUTO_RECOVERY_LOOKBACK_HOURS,
|
||||
};
|
||||
@@ -104,7 +267,7 @@ function toInstanceSettings(row: typeof instanceSettings.$inferSelect): Instance
|
||||
} as InstanceSettings;
|
||||
}
|
||||
|
||||
export function instanceSettingsService(db: Db) {
|
||||
export function instanceSettingsService(db: Db, options: InstanceSettingsServiceOptions = {}) {
|
||||
async function getOrCreateRow() {
|
||||
const existing = await db
|
||||
.select()
|
||||
@@ -192,10 +355,7 @@ export function instanceSettingsService(db: Db) {
|
||||
|
||||
updateExperimental: async (patch: PatchInstanceExperimentalSettings): Promise<InstanceSettings> => {
|
||||
const current = await getOrCreateRow();
|
||||
const nextExperimental = normalizeExperimentalSettings({
|
||||
...normalizeExperimentalSettings(current.experimental),
|
||||
...patch,
|
||||
});
|
||||
const nextExperimental = applyExperimentalSettingsPatch(current.experimental, patch, options);
|
||||
const now = new Date();
|
||||
const [updated] = await db
|
||||
.update(instanceSettings)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { and, asc, desc, eq, gt, inArray, isNull, notInArray, sql } from "drizzle-orm";
|
||||
import { and, asc, desc, eq, gt, gte, inArray, isNull, notInArray, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { clampIssueRequestDepth } from "@paperclipai/shared";
|
||||
import {
|
||||
@@ -763,6 +763,7 @@ export function productivityReviewService(db: Db, deps?: { enqueueWakeup?: Enque
|
||||
now?: Date;
|
||||
companyId?: string;
|
||||
thresholds?: Partial<ProductivityReviewThresholds>;
|
||||
issueCreatedAtGte?: Date | null;
|
||||
}) {
|
||||
const now = opts?.now ?? new Date();
|
||||
const thresholds = buildThresholds(opts?.thresholds);
|
||||
@@ -777,6 +778,7 @@ export function productivityReviewService(db: Db, deps?: { enqueueWakeup?: Enque
|
||||
inArray(issues.status, ["todo", "in_progress"]),
|
||||
sql`${issues.assigneeAgentId} is not null`,
|
||||
sql`${issues.originKind} <> ${PRODUCTIVITY_REVIEW_ORIGIN_KIND}`,
|
||||
opts?.issueCreatedAtGte ? gte(issues.createdAt, opts.issueCreatedAtGte) : undefined,
|
||||
),
|
||||
)
|
||||
.orderBy(asc(issues.updatedAt), asc(issues.id))
|
||||
|
||||
@@ -2009,10 +2009,10 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
return { kind: "created" as const, evaluationIssueId: evaluation.id };
|
||||
}
|
||||
|
||||
async function scanSilentActiveRuns(opts?: { now?: Date; companyId?: string }) {
|
||||
async function scanSilentActiveRuns(opts?: { now?: Date; companyId?: string; issueCreatedAtGte?: Date | null }) {
|
||||
const now = opts?.now ?? new Date();
|
||||
const suspicionBefore = new Date(now.getTime() - ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS);
|
||||
const candidates = await db
|
||||
let candidates = await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
@@ -2025,6 +2025,27 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
.orderBy(asc(heartbeatRuns.createdAt))
|
||||
.limit(100);
|
||||
|
||||
if (opts?.issueCreatedAtGte) {
|
||||
const issueIds = [...new Set(candidates.flatMap((run) => {
|
||||
const context = parseObject(run.contextSnapshot);
|
||||
const issueId = context.issueId ?? context.taskId;
|
||||
return typeof issueId === "string" && issueId.length > 0 ? [issueId] : [];
|
||||
}))];
|
||||
const eligibleIssueIds = new Set(
|
||||
issueIds.length > 0
|
||||
? (await db.select({ id: issues.id }).from(issues).where(and(
|
||||
inArray(issues.id, issueIds),
|
||||
gte(issues.createdAt, opts.issueCreatedAtGte),
|
||||
))).map((issue) => issue.id)
|
||||
: [],
|
||||
);
|
||||
candidates = candidates.filter((run) => {
|
||||
const context = parseObject(run.contextSnapshot);
|
||||
const issueId = context.issueId ?? context.taskId;
|
||||
return typeof issueId === "string" && eligibleIssueIds.has(issueId);
|
||||
});
|
||||
}
|
||||
|
||||
const result = {
|
||||
scanned: candidates.length,
|
||||
created: 0,
|
||||
@@ -2955,7 +2976,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
return updated;
|
||||
}
|
||||
|
||||
async function reconcileStrandedAssignedIssues() {
|
||||
async function reconcileStrandedAssignedIssues(opts?: { issueCreatedAtGte?: Date | null }) {
|
||||
const candidates = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
@@ -2967,6 +2988,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
sql`${issues.assigneeAgentId} is not null`,
|
||||
eq(issues.status, "in_review"),
|
||||
),
|
||||
opts?.issueCreatedAtGte ? gte(issues.createdAt, opts.issueCreatedAtGte) : undefined,
|
||||
),
|
||||
);
|
||||
|
||||
@@ -4488,8 +4510,25 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
runId?: string | null;
|
||||
force?: boolean;
|
||||
lookbackHours?: number;
|
||||
issueCreatedAtGte?: Date | null;
|
||||
}) {
|
||||
const findings = await collectIssueGraphLivenessFindings();
|
||||
let findings = await collectIssueGraphLivenessFindings();
|
||||
if (opts?.issueCreatedAtGte) {
|
||||
const findingIssueIds = [...new Set(findings.map((finding) => finding.recoveryIssueId))];
|
||||
const eligibleIssueIds = new Set(
|
||||
findingIssueIds.length === 0
|
||||
? []
|
||||
: (await db
|
||||
.select({ id: issues.id })
|
||||
.from(issues)
|
||||
.where(and(
|
||||
inArray(issues.id, findingIssueIds),
|
||||
gte(issues.createdAt, opts.issueCreatedAtGte),
|
||||
)))
|
||||
.map((issue) => issue.id),
|
||||
);
|
||||
findings = findings.filter((finding) => eligibleIssueIds.has(finding.recoveryIssueId));
|
||||
}
|
||||
const experimentalSettings = await instanceSettings.getExperimental();
|
||||
const autoRecoveryEnabled = asBoolean(
|
||||
experimentalSettings.enableIssueGraphLivenessAutoRecovery,
|
||||
|
||||
@@ -65,6 +65,12 @@ import { secretService } from "./secrets.js";
|
||||
import { getSecretProvider } from "../secrets/provider-registry.js";
|
||||
import { parseCron, validateCron } from "./cron.js";
|
||||
import { heartbeatService } from "./heartbeat.js";
|
||||
import {
|
||||
instanceSettingsService,
|
||||
isTruthyRuntimeEnvValue,
|
||||
resolveWorktreeRunExecutionActivationState,
|
||||
type WorktreeRunExecutionActivationState,
|
||||
} from "./instance-settings.js";
|
||||
import { queueIssueAssignmentWakeup, type IssueAssignmentWakeupDeps } from "./issue-assignment-wakeup.js";
|
||||
import { logActivity } from "./activity-log.js";
|
||||
import type { PluginWorkerManager } from "./plugin-worker-manager.js";
|
||||
@@ -592,10 +598,13 @@ export function routineService(
|
||||
deps: {
|
||||
heartbeat?: IssueAssignmentWakeupDeps;
|
||||
pluginWorkerManager?: PluginWorkerManager;
|
||||
runtimeEnv?: Record<string, string | undefined>;
|
||||
} = {},
|
||||
) {
|
||||
const issueSvc = issueService(db);
|
||||
const secretsSvc = secretService(db);
|
||||
const instanceSettings = instanceSettingsService(db);
|
||||
const runtimeEnv = deps.runtimeEnv ?? process.env;
|
||||
const heartbeat = deps.heartbeat ?? heartbeatService(db, {
|
||||
pluginWorkerManager: deps.pluginWorkerManager,
|
||||
});
|
||||
@@ -1155,14 +1164,32 @@ export function routineService(
|
||||
}
|
||||
}
|
||||
|
||||
// Records a skipped scheduled firing without creating an execution issue. Used when the
|
||||
// routine's project is paused: the tick is still claimed/advanced upstream (no backfill),
|
||||
// and run history + trigger audit reflect the pause-specific skip.
|
||||
async function recordSuppressedScheduleRun(input: {
|
||||
async function getAutomaticRoutineDispatchEligibility(
|
||||
routine: typeof routines.$inferSelect,
|
||||
activation?: WorktreeRunExecutionActivationState,
|
||||
) {
|
||||
if (!isTruthyRuntimeEnvValue(runtimeEnv.PAPERCLIP_IN_WORKTREE)) return { eligible: true };
|
||||
|
||||
const resolvedActivation = activation ?? await resolveWorktreeRunExecutionActivationState({
|
||||
getExperimental: instanceSettings.getExperimental,
|
||||
runtimeEnv,
|
||||
});
|
||||
if (!resolvedActivation.armed) return { eligible: false };
|
||||
|
||||
const cutoff = new Date(resolvedActivation.cutoff);
|
||||
if (Number.isNaN(cutoff.getTime()) || routine.createdAt < cutoff) return { eligible: false };
|
||||
return { eligible: true };
|
||||
}
|
||||
|
||||
// Records an automatic firing that was claimed but intentionally not dispatched. The
|
||||
// scheduler advances its tick before calling this helper, so suppressed work is never
|
||||
// replayed after a setting or project state changes.
|
||||
async function recordSuppressedAutomaticRun(input: {
|
||||
routine: typeof routines.$inferSelect;
|
||||
trigger: typeof routineTriggers.$inferSelect;
|
||||
source: "schedule" | "webhook";
|
||||
reason: string;
|
||||
nextRunAt: Date | null;
|
||||
nextRunAt?: Date | null;
|
||||
}) {
|
||||
const triggeredAt = new Date();
|
||||
const run = await db.transaction(async (tx) => {
|
||||
@@ -1173,7 +1200,7 @@ export function routineService(
|
||||
companyId: input.routine.companyId,
|
||||
routineId: input.routine.id,
|
||||
triggerId: input.trigger.id,
|
||||
source: "schedule",
|
||||
source: input.source,
|
||||
status: "skipped",
|
||||
triggeredAt,
|
||||
failureReason: input.reason,
|
||||
@@ -1187,7 +1214,7 @@ export function routineService(
|
||||
routineId: input.routine.id,
|
||||
triggerId: input.trigger.id,
|
||||
triggeredAt,
|
||||
status: "skipped_paused",
|
||||
status: input.reason === "paused" ? "skipped_paused" : "skipped_worktree_execution_cutoff",
|
||||
nextRunAt: input.nextRunAt,
|
||||
}, txDb);
|
||||
return createdRun;
|
||||
@@ -1197,14 +1224,14 @@ export function routineService(
|
||||
await logActivity(db, {
|
||||
companyId: input.routine.companyId,
|
||||
actorType: "system",
|
||||
actorId: "routine-scheduler",
|
||||
actorId: input.source === "schedule" ? "routine-scheduler" : "routine-webhook",
|
||||
action: "routine.run_skipped",
|
||||
entityType: "routine_run",
|
||||
entityId: run.id,
|
||||
details: {
|
||||
routineId: input.routine.id,
|
||||
triggerId: input.trigger.id,
|
||||
source: "schedule",
|
||||
source: input.source,
|
||||
status: "skipped",
|
||||
reason: input.reason,
|
||||
},
|
||||
@@ -2645,6 +2672,16 @@ export function routineService(
|
||||
if (!valid) throw unauthorized();
|
||||
}
|
||||
|
||||
const eligibility = await getAutomaticRoutineDispatchEligibility(routine);
|
||||
if (!eligibility.eligible) {
|
||||
return recordSuppressedAutomaticRun({
|
||||
routine,
|
||||
trigger,
|
||||
source: "webhook",
|
||||
reason: "worktree_execution_cutoff",
|
||||
});
|
||||
}
|
||||
|
||||
return dispatchRoutineRun({
|
||||
routine,
|
||||
trigger,
|
||||
@@ -2732,6 +2769,12 @@ export function routineService(
|
||||
},
|
||||
|
||||
tickScheduledTriggers: async (now: Date = new Date()) => {
|
||||
const worktreeActivation = isTruthyRuntimeEnvValue(runtimeEnv.PAPERCLIP_IN_WORKTREE)
|
||||
? await resolveWorktreeRunExecutionActivationState({
|
||||
getExperimental: instanceSettings.getExperimental,
|
||||
runtimeEnv,
|
||||
})
|
||||
: undefined;
|
||||
const due = await db
|
||||
.select({
|
||||
trigger: routineTriggers,
|
||||
@@ -2761,11 +2804,13 @@ export function routineService(
|
||||
// at the next cron boundary instead of replaying missed firings. Routines with no
|
||||
// project are never suppressed here.
|
||||
const projectPaused = !!(row.routine.projectId && row.projectPausedAt);
|
||||
const automaticEligibility = await getAutomaticRoutineDispatchEligibility(row.routine, worktreeActivation);
|
||||
const worktreeSuppressed = !automaticEligibility.eligible;
|
||||
|
||||
let runCount = 1;
|
||||
let claimedNextRunAt = nextCronTickInTimeZone(row.trigger.cronExpression, row.trigger.timezone, now);
|
||||
|
||||
if (!projectPaused && row.routine.catchUpPolicy === "enqueue_missed_with_cap") {
|
||||
if (!projectPaused && !worktreeSuppressed && row.routine.catchUpPolicy === "enqueue_missed_with_cap") {
|
||||
let cursor: Date | null = row.trigger.nextRunAt;
|
||||
runCount = 0;
|
||||
while (cursor && cursor <= now && runCount < MAX_CATCH_UP_RUNS) {
|
||||
@@ -2792,11 +2837,12 @@ export function routineService(
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!claimed) continue;
|
||||
|
||||
if (projectPaused) {
|
||||
await recordSuppressedScheduleRun({
|
||||
if (projectPaused || worktreeSuppressed) {
|
||||
await recordSuppressedAutomaticRun({
|
||||
routine: row.routine,
|
||||
trigger: row.trigger,
|
||||
reason: "paused",
|
||||
source: "schedule",
|
||||
reason: worktreeSuppressed ? "worktree_execution_cutoff" : "paused",
|
||||
nextRunAt: claimedNextRunAt,
|
||||
});
|
||||
continue;
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { and, asc, eq, inArray, isNull, or, sql } from "drizzle-orm";
|
||||
import { and, asc, eq, gte, inArray, isNull, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
agentWakeupRequests,
|
||||
@@ -1560,8 +1560,28 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
|
||||
return toIssueWatchdog(updated);
|
||||
},
|
||||
|
||||
reconcileTaskWatchdogs: async (opts: { companyId?: string | null; runId?: string | null } = {}) => {
|
||||
const rows = await listActiveWatchdogsForCompany(opts.companyId ?? null);
|
||||
reconcileTaskWatchdogs: async (opts: {
|
||||
companyId?: string | null;
|
||||
runId?: string | null;
|
||||
issueCreatedAtGte?: Date | null;
|
||||
} = {}) => {
|
||||
let rows = await listActiveWatchdogsForCompany(opts.companyId ?? null);
|
||||
if (opts.issueCreatedAtGte) {
|
||||
const watchdogIssueIds = [...new Set(rows.map((row) => row.issueId))];
|
||||
const eligibleIssueIds = new Set(
|
||||
watchdogIssueIds.length === 0
|
||||
? []
|
||||
: (await db
|
||||
.select({ id: issues.id })
|
||||
.from(issues)
|
||||
.where(and(
|
||||
inArray(issues.id, watchdogIssueIds),
|
||||
gte(issues.createdAt, opts.issueCreatedAtGte),
|
||||
)))
|
||||
.map((issue) => issue.id),
|
||||
);
|
||||
rows = rows.filter((row) => eligibleIssueIds.has(row.issueId));
|
||||
}
|
||||
const result = {
|
||||
checked: 0,
|
||||
triggered: 0,
|
||||
|
||||
Reference in new issue
Block a user