feat(routines): gate scheduled runs on external activity (#9436)
## Thinking Path > - Paperclip is the open source app people use to manage AI-agent companies and their recurring work > - Scheduled routines provide native cron-driven execution for recurring agent tasks > - Watcher-style routines currently dispatch a model run even when the control plane has been quiet since their last useful run > - Existing pause, catch-up, and concurrency policies do not distinguish external work from a routine's own bookkeeping > - This pull request adds a generic activity gate that checks company-scoped activity provenance before scheduled dispatch > - The benefit is backward-compatible zero-token quiet skips while real human, agent, or delegated-child activity still wakes the routine ## Linked Issues or Issue Description - Refs #8534 ## What Changed - Added `activity_gate_policy` and `activity_gate_scope` routine columns with backward-compatible `always` / `company` defaults. - Added a company-bounded `evaluateActivityGate()` predicate that uses the last dispatched run as its open window, excludes the routine's own execution runs and scheduler bookkeeping, ignores pure-read actions, and supports company/project scope. - Integrated the predicate into scheduled ticks after pause/worktree eligibility checks; quiet ticks create visible skipped run-history rows with reason `no_external_activity` and gate-window diagnostics without advancing the activity window. - Kept webhook, manual, and API dispatch paths ungated; catch-up schedules evaluate the gate once per scheduler tick. - Added migration-default, provenance predicate, project-scope, quiet-window, scheduler, and webhook-bypass coverage. ## Verification - `pnpm --filter @paperclipai/db typecheck` - `pnpm --filter @paperclipai/server typecheck` - `pnpm exec vitest run server/src/__tests__/routines-service.test.ts` — 51 tests passed - Embedded Postgres `EXPLAIN` for the company-scope gate scan: ```text Limit (cost=24.56..24.58 rows=1 width=24) -> Incremental Sort (cost=24.56..24.60 rows=2 width=24) Sort Key: activity.created_at, activity.id Presorted Key: activity.created_at -> Nested Loop Anti Join (cost=0.44..24.55 rows=1 width=24) Join Filter: (own_run.id = activity.run_id) -> Index Scan using activity_log_company_created_idx on activity_log activity (cost=0.15..8.19 rows=1 width=40) Index Cond: ((company_id = '00000000-0000-0000-0000-000000000001'::uuid) AND (created_at > (now() - '01:00:00'::interval)) AND (created_at <= now())) ``` ## Risks - The migration adds two non-null text columns, but constant defaults preserve all existing routine behavior and avoid a backfill step. - Project scope resolves activity through issue/run/routine provenance; tests cover in-project and cross-project issue activity, while every top-level and correlated query remains company-bounded. - This is the scheduler/schema foundation. Public API validation and documentation for configuring the new fields are intentionally handled in the next scoped follow-up. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex using `gpt-5.4` with medium reasoning, repository/tool access, terminal code execution, and test execution. The runtime did not expose a context-window size. ## 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 extends the existing Scheduled Routines roadmap item - [x] I have searched GitHub for duplicate or related PRs and linked the related efficiency request 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) - [x] My branch name describes the change 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 (no user-facing configuration is exposed in this scoped foundation PR) - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] 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>
This commit is contained in:
Dotta
GitHub
parent
e4e12bfb89
commit
4a40c0cb13
|
|
@ -0,0 +1,3 @@
|
|||
ALTER TABLE "routines" ADD COLUMN "activity_gate_policy" text DEFAULT 'always' NOT NULL;
|
||||
--> statement-breakpoint
|
||||
ALTER TABLE "routines" ADD COLUMN "activity_gate_scope" text DEFAULT 'company' NOT NULL;
|
||||
|
|
@ -1009,6 +1009,13 @@
|
|||
"when": 1783641600000,
|
||||
"tag": "0145_inbox_dismissal_snooze_kind",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 146,
|
||||
"version": "7",
|
||||
"when": 1783822632557,
|
||||
"tag": "0146_routine_activity_gate",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -34,6 +34,8 @@ export const routines = pgTable(
|
|||
status: text("status").notNull().default("active"),
|
||||
concurrencyPolicy: text("concurrency_policy").notNull().default("coalesce_if_active"),
|
||||
catchUpPolicy: text("catch_up_policy").notNull().default("skip_missed"),
|
||||
activityGatePolicy: text("activity_gate_policy").notNull().default("always"),
|
||||
activityGateScope: text("activity_gate_scope").notNull().default("company"),
|
||||
originKind: text("origin_kind").notNull().default("manual"),
|
||||
originId: text("origin_id"),
|
||||
variables: jsonb("variables").$type<RoutineVariable[]>().notNull().default([]),
|
||||
|
|
|
|||
|
|
@ -218,6 +218,26 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => {
|
|||
});
|
||||
}
|
||||
|
||||
async function insertDispatchedRun(input: {
|
||||
companyId: string;
|
||||
routineId: string;
|
||||
triggeredAt: Date;
|
||||
source?: "schedule" | "manual" | "api" | "webhook";
|
||||
}) {
|
||||
return db
|
||||
.insert(routineRuns)
|
||||
.values({
|
||||
companyId: input.companyId,
|
||||
routineId: input.routineId,
|
||||
source: input.source ?? "schedule",
|
||||
status: "completed",
|
||||
triggeredAt: input.triggeredAt,
|
||||
completedAt: input.triggeredAt,
|
||||
})
|
||||
.returning()
|
||||
.then((rows) => rows[0]!);
|
||||
}
|
||||
|
||||
it("filters listed routines by project", async () => {
|
||||
const { companyId, agentId, projectId, routine, svc } = await seedFixture();
|
||||
const otherProjectId = randomUUID();
|
||||
|
|
@ -251,6 +271,223 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => {
|
|||
expect(allRoutines.map((entry) => entry.id)).toEqual(expect.arrayContaining([routine.id, otherRoutine.id]));
|
||||
});
|
||||
|
||||
it("defaults activity gates to always at company scope", async () => {
|
||||
const { routine } = await seedFixture();
|
||||
|
||||
expect(routine.activityGatePolicy).toBe("always");
|
||||
expect(routine.activityGateScope).toBe("company");
|
||||
});
|
||||
|
||||
it("fires an activity gate for a routine that has never dispatched", async () => {
|
||||
const { routine, svc } = await seedFixture();
|
||||
|
||||
await expect(svc.evaluateActivityGate(routine, new Date())).resolves.toEqual({
|
||||
fire: true,
|
||||
windowStart: null,
|
||||
matchedActivity: null,
|
||||
});
|
||||
});
|
||||
|
||||
it("excludes activity from heartbeat runs executing the routine's own issue", async () => {
|
||||
const { agentId, companyId, projectId, routine, svc } = await seedFixture();
|
||||
const windowStart = new Date(Date.now() - 60_000);
|
||||
const now = new Date();
|
||||
await insertDispatchedRun({ companyId, routineId: routine.id, triggeredAt: windowStart });
|
||||
const issueId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
projectId,
|
||||
title: "Routine execution",
|
||||
originKind: "routine_execution",
|
||||
originId: routine.id,
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "completed",
|
||||
contextSnapshot: { issueId },
|
||||
});
|
||||
await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "agent",
|
||||
actorId: agentId,
|
||||
agentId,
|
||||
runId,
|
||||
action: "issue.comment_added",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
createdAt: new Date(windowStart.getTime() + 1_000),
|
||||
});
|
||||
|
||||
await expect(svc.evaluateActivityGate(routine, now)).resolves.toMatchObject({
|
||||
fire: false,
|
||||
windowStart,
|
||||
matchedActivity: null,
|
||||
});
|
||||
});
|
||||
|
||||
it("fires for another agent running a child of the routine issue", async () => {
|
||||
const { agentId, companyId, projectId, routine, svc } = await seedFixture();
|
||||
const windowStart = new Date(Date.now() - 60_000);
|
||||
const now = new Date();
|
||||
await insertDispatchedRun({ companyId, routineId: routine.id, triggeredAt: windowStart });
|
||||
const otherAgentId = randomUUID();
|
||||
await db.insert(agents).values({
|
||||
id: otherAgentId,
|
||||
companyId,
|
||||
name: "Worker",
|
||||
role: "engineer",
|
||||
status: "active",
|
||||
adapterType: "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
const routineIssueId = randomUUID();
|
||||
const childIssueId = randomUUID();
|
||||
await db.insert(issues).values([
|
||||
{
|
||||
id: routineIssueId,
|
||||
companyId,
|
||||
projectId,
|
||||
title: "Routine execution",
|
||||
originKind: "routine_execution",
|
||||
originId: routine.id,
|
||||
},
|
||||
{
|
||||
id: childIssueId,
|
||||
companyId,
|
||||
projectId,
|
||||
parentId: routineIssueId,
|
||||
title: "Delegated child",
|
||||
},
|
||||
]);
|
||||
const childRunId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: childRunId,
|
||||
companyId,
|
||||
agentId: otherAgentId,
|
||||
status: "running",
|
||||
contextSnapshot: { issueId: childIssueId },
|
||||
});
|
||||
const [activity] = await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "agent",
|
||||
actorId: otherAgentId,
|
||||
agentId: otherAgentId,
|
||||
runId: childRunId,
|
||||
action: "issue.checkout",
|
||||
entityType: "issue",
|
||||
entityId: childIssueId,
|
||||
createdAt: new Date(windowStart.getTime() + 1_000),
|
||||
}).returning();
|
||||
|
||||
await expect(svc.evaluateActivityGate(routine, now)).resolves.toMatchObject({
|
||||
fire: true,
|
||||
windowStart,
|
||||
matchedActivity: { id: activity!.id },
|
||||
});
|
||||
expect(agentId).not.toBe(otherAgentId);
|
||||
});
|
||||
|
||||
it("fires for a human comment and ignores pure-read activity", async () => {
|
||||
const { companyId, projectId, routine, svc } = await seedFixture();
|
||||
const windowStart = new Date(Date.now() - 60_000);
|
||||
const now = new Date();
|
||||
await insertDispatchedRun({ companyId, routineId: routine.id, triggeredAt: windowStart });
|
||||
const issueId = randomUUID();
|
||||
await db.insert(issues).values({ id: issueId, companyId, projectId, title: "Board task" });
|
||||
await db.insert(activityLog).values([
|
||||
{
|
||||
companyId,
|
||||
actorType: "user",
|
||||
actorId: "user-1",
|
||||
action: "issue.read_marked",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
createdAt: new Date(windowStart.getTime() + 1_000),
|
||||
},
|
||||
{
|
||||
companyId,
|
||||
actorType: "user",
|
||||
actorId: "user-1",
|
||||
action: "issue.comment_added",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
createdAt: new Date(windowStart.getTime() + 2_000),
|
||||
},
|
||||
]);
|
||||
|
||||
await expect(svc.evaluateActivityGate(routine, now)).resolves.toMatchObject({
|
||||
fire: true,
|
||||
matchedActivity: { action: "issue.comment_added" },
|
||||
});
|
||||
|
||||
await db.delete(activityLog);
|
||||
await db.insert(activityLog).values([
|
||||
{
|
||||
companyId,
|
||||
actorType: "user",
|
||||
actorId: "user-1",
|
||||
action: "issue.read_marked",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
createdAt: new Date(windowStart.getTime() + 1_000),
|
||||
},
|
||||
{
|
||||
companyId,
|
||||
actorType: "user",
|
||||
actorId: "user-1",
|
||||
action: "issue.inbox_archived",
|
||||
entityType: "issue",
|
||||
entityId: issueId,
|
||||
createdAt: new Date(windowStart.getTime() + 2_000),
|
||||
},
|
||||
]);
|
||||
|
||||
await expect(svc.evaluateActivityGate(routine, now)).resolves.toMatchObject({ fire: false });
|
||||
});
|
||||
|
||||
it("limits project-scoped gates to activity in the routine project", async () => {
|
||||
const { companyId, projectId, routine, svc } = await seedFixture();
|
||||
const otherProjectId = randomUUID();
|
||||
await db.insert(projects).values({ id: otherProjectId, companyId, name: "Other", status: "in_progress" });
|
||||
const windowStart = new Date(Date.now() - 60_000);
|
||||
const now = new Date();
|
||||
await insertDispatchedRun({ companyId, routineId: routine.id, triggeredAt: windowStart });
|
||||
const [otherIssue, ownIssue] = [randomUUID(), randomUUID()];
|
||||
await db.insert(issues).values([
|
||||
{ id: otherIssue, companyId, projectId: otherProjectId, title: "Other project" },
|
||||
{ id: ownIssue, companyId, projectId, title: "Routine project" },
|
||||
]);
|
||||
const projectRoutine = { ...routine, activityGateScope: "project" };
|
||||
await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "user",
|
||||
actorId: "user-1",
|
||||
action: "issue.comment_added",
|
||||
entityType: "issue",
|
||||
entityId: otherIssue,
|
||||
createdAt: new Date(windowStart.getTime() + 1_000),
|
||||
});
|
||||
|
||||
await expect(svc.evaluateActivityGate(projectRoutine, now)).resolves.toMatchObject({ fire: false });
|
||||
|
||||
await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "user",
|
||||
actorId: "user-1",
|
||||
action: "issue.comment_added",
|
||||
entityType: "issue",
|
||||
entityId: ownIssue,
|
||||
createdAt: new Date(windowStart.getTime() + 2_000),
|
||||
});
|
||||
await expect(svc.evaluateActivityGate(projectRoutine, now)).resolves.toMatchObject({ fire: true });
|
||||
});
|
||||
|
||||
it("creates a fresh execution issue when the previous routine issue is open but idle", async () => {
|
||||
const { companyId, issueSvc, routine, svc } = await seedFixture();
|
||||
const previousRunId = randomUUID();
|
||||
|
|
@ -1917,4 +2154,72 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => {
|
|||
expect(runsAfterResume).toHaveLength(2);
|
||||
expect(runsAfterResume.some((run) => run.status === "issue_created")).toBe(true);
|
||||
});
|
||||
|
||||
it("skips a gated scheduled tick when quiet without advancing the activity window", async () => {
|
||||
const { companyId, routine, svc } = await seedFixture();
|
||||
await db.update(routines).set({
|
||||
activityGatePolicy: "require_external_activity",
|
||||
}).where(eq(routines.id, routine.id));
|
||||
const gatedRoutine = { ...routine, activityGatePolicy: "require_external_activity" };
|
||||
const { trigger } = await svc.createTrigger(routine.id, {
|
||||
kind: "schedule",
|
||||
cronExpression: "* * * * *",
|
||||
timezone: "UTC",
|
||||
}, {});
|
||||
const firstTick = new Date();
|
||||
await db.update(routineTriggers).set({ nextRunAt: new Date(firstTick.getTime() - 1_000) }).where(eq(routineTriggers.id, trigger.id));
|
||||
|
||||
expect(await svc.tickScheduledTriggers(firstTick)).toEqual({ triggered: 1 });
|
||||
const [firstRun] = await db.select().from(routineRuns).where(eq(routineRuns.routineId, routine.id));
|
||||
expect(firstRun?.status).toBe("issue_created");
|
||||
|
||||
const quietTick = new Date(firstTick.getTime() + 60_000);
|
||||
await db.update(routineTriggers).set({ nextRunAt: new Date(quietTick.getTime() - 1_000) }).where(eq(routineTriggers.id, trigger.id));
|
||||
expect(await svc.tickScheduledTriggers(quietTick)).toEqual({ triggered: 0 });
|
||||
|
||||
const runsAfterQuietTick = await db.select().from(routineRuns).where(eq(routineRuns.routineId, routine.id));
|
||||
const quietRun = runsAfterQuietTick.find((run) => run.failureReason === "no_external_activity");
|
||||
expect(quietRun).toMatchObject({
|
||||
status: "skipped",
|
||||
source: "schedule",
|
||||
linkedIssueId: null,
|
||||
triggerPayload: {
|
||||
activityGate: {
|
||||
verdict: "quiet",
|
||||
windowStart: firstRun!.triggeredAt.toISOString(),
|
||||
matchedActivityId: null,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const activityAt = new Date(firstRun!.triggeredAt.getTime() + 30_000);
|
||||
await db.insert(activityLog).values({
|
||||
companyId,
|
||||
actorType: "user",
|
||||
actorId: "user-1",
|
||||
action: "issue.comment_added",
|
||||
entityType: "issue",
|
||||
entityId: firstRun!.linkedIssueId!,
|
||||
createdAt: activityAt,
|
||||
});
|
||||
await db.update(issues).set({ status: "done", completedAt: activityAt }).where(eq(issues.id, firstRun!.linkedIssueId!));
|
||||
const resumedTick = new Date(quietTick.getTime() + 60_000);
|
||||
await db.update(routineTriggers).set({ nextRunAt: new Date(resumedTick.getTime() - 1_000) }).where(eq(routineTriggers.id, trigger.id));
|
||||
|
||||
await expect(svc.evaluateActivityGate(gatedRoutine, resumedTick)).resolves.toMatchObject({
|
||||
fire: true,
|
||||
windowStart: firstRun!.triggeredAt,
|
||||
});
|
||||
expect(await svc.tickScheduledTriggers(resumedTick)).toEqual({ triggered: 1 });
|
||||
});
|
||||
|
||||
it("bypasses the activity gate for webhook dispatches", async () => {
|
||||
const { routine, svc } = await seedFixture();
|
||||
await db.update(routines).set({ activityGatePolicy: "require_external_activity" }).where(eq(routines.id, routine.id));
|
||||
const { trigger } = await svc.createTrigger(routine.id, { kind: "webhook", signingMode: "none" }, {});
|
||||
|
||||
const run = await svc.firePublicTrigger(trigger.publicId!, { payload: { source: "test" } });
|
||||
|
||||
expect(run).toMatchObject({ source: "webhook", status: "issue_created" });
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,8 +1,9 @@
|
|||
import crypto from "node:crypto";
|
||||
import { and, asc, desc, eq, inArray, isNotNull, isNull, lte, ne, not, or, sql } from "drizzle-orm";
|
||||
import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lte, ne, not, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
agents,
|
||||
activityLog,
|
||||
companies,
|
||||
companyMemberships,
|
||||
companySecretBindings,
|
||||
|
|
@ -80,6 +81,12 @@ const LIVE_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"];
|
|||
const TERMINAL_ISSUE_STATUSES = new Set(["done", "cancelled"]);
|
||||
const MAX_CATCH_UP_RUNS = 25;
|
||||
const MAX_ROUTINE_REVISIONS = 100;
|
||||
const ACTIVITY_GATE_IGNORED_ACTIONS = [
|
||||
"issue.read_marked",
|
||||
"issue.read_unmarked",
|
||||
"issue.inbox_archived",
|
||||
"issue.inbox_unarchived",
|
||||
];
|
||||
const WEEKDAY_INDEX: Record<string, number> = {
|
||||
Sun: 0,
|
||||
Mon: 1,
|
||||
|
|
@ -1181,6 +1188,116 @@ export function routineService(
|
|||
return { eligible: true };
|
||||
}
|
||||
|
||||
async function evaluateActivityGate(routine: typeof routines.$inferSelect, now: Date) {
|
||||
const lastDispatchedRun = await db
|
||||
.select({ triggeredAt: routineRuns.triggeredAt })
|
||||
.from(routineRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(routineRuns.companyId, routine.companyId),
|
||||
eq(routineRuns.routineId, routine.id),
|
||||
sql`${routineRuns.status} not in ('skipped', 'coalesced')`,
|
||||
),
|
||||
)
|
||||
.orderBy(desc(routineRuns.triggeredAt), desc(routineRuns.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (!lastDispatchedRun) {
|
||||
return { fire: true, windowStart: null, matchedActivity: null };
|
||||
}
|
||||
|
||||
const projectScopeCondition = routine.activityGateScope === "project"
|
||||
? routine.projectId
|
||||
? sql`(
|
||||
(${activityLog.entityType} = 'project' and ${activityLog.entityId} = ${routine.projectId})
|
||||
or (${activityLog.details} ->> 'projectId') = ${routine.projectId}
|
||||
or exists (
|
||||
select 1
|
||||
from ${issues} activity_issue
|
||||
where activity_issue.company_id = ${routine.companyId}
|
||||
and activity_issue.project_id = ${routine.projectId}
|
||||
and activity_issue.id::text = ${activityLog.entityId}
|
||||
and ${activityLog.entityType} = 'issue'
|
||||
)
|
||||
or exists (
|
||||
select 1
|
||||
from ${heartbeatRuns} activity_run
|
||||
inner join ${issues} run_issue
|
||||
on run_issue.company_id = ${routine.companyId}
|
||||
and run_issue.id::text = activity_run.context_snapshot ->> 'issueId'
|
||||
where activity_run.company_id = ${routine.companyId}
|
||||
and activity_run.id = ${activityLog.runId}
|
||||
and run_issue.project_id = ${routine.projectId}
|
||||
)
|
||||
or exists (
|
||||
select 1
|
||||
from ${routines} activity_routine
|
||||
where activity_routine.company_id = ${routine.companyId}
|
||||
and activity_routine.project_id = ${routine.projectId}
|
||||
and activity_routine.id::text = ${activityLog.entityId}
|
||||
and ${activityLog.entityType} = 'routine'
|
||||
)
|
||||
or exists (
|
||||
select 1
|
||||
from ${routineRuns} activity_routine_run
|
||||
inner join ${routines} activity_routine
|
||||
on activity_routine.company_id = ${routine.companyId}
|
||||
and activity_routine.id = activity_routine_run.routine_id
|
||||
where activity_routine_run.company_id = ${routine.companyId}
|
||||
and activity_routine_run.id::text = ${activityLog.entityId}
|
||||
and activity_routine.project_id = ${routine.projectId}
|
||||
and ${activityLog.entityType} = 'routine_run'
|
||||
)
|
||||
)`
|
||||
: sql`false`
|
||||
: undefined;
|
||||
|
||||
const matchedActivity = await db
|
||||
.select({
|
||||
id: activityLog.id,
|
||||
action: activityLog.action,
|
||||
createdAt: activityLog.createdAt,
|
||||
})
|
||||
.from(activityLog)
|
||||
.where(
|
||||
and(
|
||||
eq(activityLog.companyId, routine.companyId),
|
||||
gt(activityLog.createdAt, lastDispatchedRun.triggeredAt),
|
||||
lte(activityLog.createdAt, now),
|
||||
sql`${activityLog.action} not in (${sql.join(ACTIVITY_GATE_IGNORED_ACTIONS.map((action) => sql`${action}`), sql`, `)})`,
|
||||
sql`not (
|
||||
${activityLog.actorId} = 'routine-scheduler'
|
||||
and (
|
||||
(${activityLog.details} ->> 'routineId') = ${routine.id}
|
||||
or (${activityLog.entityType} = 'routine' and ${activityLog.entityId} = ${routine.id})
|
||||
)
|
||||
)`,
|
||||
sql`not exists (
|
||||
select 1
|
||||
from ${heartbeatRuns} own_run
|
||||
inner join ${issues} own_issue
|
||||
on own_issue.company_id = ${routine.companyId}
|
||||
and own_issue.id::text = own_run.context_snapshot ->> 'issueId'
|
||||
where own_run.company_id = ${routine.companyId}
|
||||
and own_run.id = ${activityLog.runId}
|
||||
and own_issue.origin_kind = 'routine_execution'
|
||||
and own_issue.origin_id = ${routine.id}
|
||||
)`,
|
||||
projectScopeCondition,
|
||||
),
|
||||
)
|
||||
.orderBy(asc(activityLog.createdAt), asc(activityLog.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
return {
|
||||
fire: matchedActivity !== null,
|
||||
windowStart: lastDispatchedRun.triggeredAt,
|
||||
matchedActivity,
|
||||
};
|
||||
}
|
||||
|
||||
// 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.
|
||||
|
|
@ -1190,6 +1307,7 @@ export function routineService(
|
|||
source: "schedule" | "webhook";
|
||||
reason: string;
|
||||
nextRunAt?: Date | null;
|
||||
details?: Record<string, unknown> | null;
|
||||
}) {
|
||||
const triggeredAt = new Date();
|
||||
const run = await db.transaction(async (tx) => {
|
||||
|
|
@ -1208,13 +1326,18 @@ export function routineService(
|
|||
linkedIssueId: null,
|
||||
routineRevisionId: input.routine.latestRevisionId,
|
||||
responsibleUserId: input.routine.responsibleUserId ?? null,
|
||||
triggerPayload: input.details ?? null,
|
||||
})
|
||||
.returning();
|
||||
await updateRoutineTouchedState({
|
||||
routineId: input.routine.id,
|
||||
triggerId: input.trigger.id,
|
||||
triggeredAt,
|
||||
status: input.reason === "paused" ? "skipped_paused" : "skipped_worktree_execution_cutoff",
|
||||
status: input.reason === "paused"
|
||||
? "skipped_paused"
|
||||
: input.reason === "no_external_activity"
|
||||
? "skipped_no_activity"
|
||||
: "skipped_worktree_execution_cutoff",
|
||||
nextRunAt: input.nextRunAt,
|
||||
}, txDb);
|
||||
return createdRun;
|
||||
|
|
@ -1234,6 +1357,7 @@ export function routineService(
|
|||
source: input.source,
|
||||
status: "skipped",
|
||||
reason: input.reason,
|
||||
...(input.details ?? {}),
|
||||
},
|
||||
});
|
||||
} catch (err) {
|
||||
|
|
@ -1776,6 +1900,7 @@ export function routineService(
|
|||
}
|
||||
|
||||
return {
|
||||
evaluateActivityGate,
|
||||
get: getRoutineById,
|
||||
getTrigger: getTriggerById,
|
||||
|
||||
|
|
@ -2848,6 +2973,27 @@ export function routineService(
|
|||
continue;
|
||||
}
|
||||
|
||||
const activityGate = row.routine.activityGatePolicy === "require_external_activity"
|
||||
? await evaluateActivityGate(row.routine, now)
|
||||
: null;
|
||||
if (activityGate && !activityGate.fire) {
|
||||
await recordSuppressedAutomaticRun({
|
||||
routine: row.routine,
|
||||
trigger: row.trigger,
|
||||
source: "schedule",
|
||||
reason: "no_external_activity",
|
||||
nextRunAt: claimedNextRunAt,
|
||||
details: {
|
||||
activityGate: {
|
||||
verdict: "quiet",
|
||||
windowStart: activityGate.windowStart?.toISOString() ?? null,
|
||||
matchedActivityId: null,
|
||||
},
|
||||
},
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
for (let i = 0; i < runCount; i += 1) {
|
||||
await dispatchRoutineRun({
|
||||
routine: row.routine,
|
||||
|
|
|
|||
Loading…
Reference in New Issue