Skip to content

Commit de319ad

Browse files
thribhuvan003thribhuvan003
committed
fix: unblock mark-sent and reclaim stuck agent jobs
Prep-status mark-sent + L3 prior-proof on transition_case, reclaim zombie running jobs every process cycle, and requeue dead/stale enqueue keys.
1 parent edbda46 commit de319ad

6 files changed

Lines changed: 279 additions & 22 deletions

File tree

‎lib/jobs/enqueue.ts‎

Lines changed: 45 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,11 @@ import { createAdminClient } from '@/lib/supabase/admin';
44
import type { Database, Json } from '@/supabase/database.types';
55

66
type AgentRole = Database['public']['Enums']['agent_role'];
7+
type JobStatus = Database['public']['Enums']['job_status'] | string;
8+
9+
const FRESH = new Set(['pending', 'completed']);
10+
/** Stuck mid-run longer than this may be reopened (deploy kill / hang). */
11+
const STALE_RUNNING_MS = 15 * 60_000;
712

813
export type EnqueueInput = {
914
case_id: string;
@@ -18,23 +23,53 @@ export type EnqueueResult = {
1823
enqueued: boolean;
1924
duplicate?: boolean;
2025
job_id?: string;
26+
requeued?: boolean;
2127
};
2228

2329
/**
2430
* Idempotent agent job enqueue — UNIQUE on idempotency_key.
31+
* Terminal rows + stale running are reopened instead of blocking forever.
2532
* @see docs/BUILD_SPEC_LOOPS.md §8.2
2633
*/
2734
export async function enqueueAgentJob(input: EnqueueInput): Promise<EnqueueResult> {
2835
const supabase = createAdminClient();
36+
const when = input.scheduled_at ?? new Date().toISOString();
2937

3038
const { data: existing } = await supabase
3139
.from('agent_jobs')
32-
.select('id')
40+
.select('id, status, started_at')
3341
.eq('idempotency_key', input.idempotency_key)
3442
.maybeSingle();
3543

3644
if (existing?.id) {
37-
return { enqueued: false, duplicate: true, job_id: existing.id };
45+
const status = existing.status as JobStatus;
46+
const startedMs = existing.started_at
47+
? Date.parse(existing.started_at)
48+
: 0;
49+
const staleRunning =
50+
status === 'running' &&
51+
startedMs > 0 &&
52+
Date.now() - startedMs >= STALE_RUNNING_MS;
53+
54+
if (FRESH.has(status) || (status === 'running' && !staleRunning)) {
55+
return { enqueued: false, duplicate: true, job_id: existing.id };
56+
}
57+
58+
const { error } = await supabase
59+
.from('agent_jobs')
60+
.update({
61+
status: 'pending',
62+
scheduled_at: when,
63+
started_at: null,
64+
completed_at: null,
65+
error_message: staleRunning ? 'requeued_stale_running' : null,
66+
payload_json: (input.payload ?? {}) as Json,
67+
})
68+
.eq('id', existing.id);
69+
70+
if (error) throw new Error(`requeue_agent_job_failed: ${error.message}`);
71+
await kickSoon();
72+
return { enqueued: true, requeued: true, job_id: existing.id };
3873
}
3974

4075
const { data, error } = await supabase
@@ -46,7 +81,7 @@ export async function enqueueAgentJob(input: EnqueueInput): Promise<EnqueueResul
4681
idempotency_key: input.idempotency_key,
4782
payload_json: (input.payload ?? {}) as Json,
4883
status: 'pending',
49-
scheduled_at: input.scheduled_at ?? new Date().toISOString(),
84+
scheduled_at: when,
5085
})
5186
.select('id')
5287
.single();
@@ -58,14 +93,15 @@ export async function enqueueAgentJob(input: EnqueueInput): Promise<EnqueueResul
5893
throw new Error(`enqueue_agent_job_failed: ${error.message}`);
5994
}
6095

61-
// Do not wait for the Hobby 10-minute cron when a user just queued work.
62-
// Kick is single-flight + best-effort; processAgentJobs stays the source of truth.
96+
await kickSoon();
97+
return { enqueued: true, job_id: data.id };
98+
}
99+
100+
async function kickSoon() {
63101
try {
64102
const { scheduleJobKick } = await import('@/lib/jobs/kick');
65103
scheduleJobKick(8);
66104
} catch {
67-
// Non-fatal — cron still drains.
105+
// Cron still drains.
68106
}
69-
70-
return { enqueued: true, job_id: data.id };
71-
}
107+
}

‎lib/jobs/kick.ts‎

Lines changed: 9 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,14 @@ import "server-only";
1010
* the in-process path is busy/unavailable — agents stay alive)
1111
*/
1212

13-
let inFlight: Promise<{ processed: number; succeeded: number; failed: number } | null> | null =
14-
null;
13+
type KickResult = {
14+
processed: number;
15+
succeeded: number;
16+
failed: number;
17+
reclaimed?: number;
18+
};
19+
20+
let inFlight: Promise<KickResult | null> | null = null;
1521

1622
function appBaseUrl(): string | null {
1723
const raw =
@@ -49,11 +55,7 @@ export async function httpKickJobWorker(limit = 8): Promise<boolean> {
4955
}
5056
}
5157

52-
export async function kickPendingJobs(limit = 8): Promise<{
53-
processed: number;
54-
succeeded: number;
55-
failed: number;
56-
} | null> {
58+
export async function kickPendingJobs(limit = 8): Promise<KickResult | null> {
5759
if (inFlight) return inFlight;
5860

5961
inFlight = (async () => {

‎lib/jobs/process.ts‎

Lines changed: 40 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@ const DEFAULT_BATCH_SIZE = 10;
1010
const MAX_ATTEMPTS = 3;
1111
const RETRY_BASE_DELAY_MS = 60_000;
1212
const RETRY_MAX_DELAY_MS = 30 * 60_000;
13+
/** Worker crashes leave status=running forever; reclaim after this. */
14+
const STALE_RUNNING_MS = 15 * 60_000;
1315

1416
type JobFailureTransition = Pick<
1517
AgentJobRow,
@@ -47,16 +49,22 @@ export function getJobFailureTransition(input: {
4749
*/
4850
export async function processAgentJobs(options?: {
4951
limit?: number;
50-
}): Promise<{ processed: number; succeeded: number; failed: number }> {
52+
}): Promise<{
53+
processed: number;
54+
succeeded: number;
55+
failed: number;
56+
reclaimed: number;
57+
}> {
5158
const supabase = createAdminClient();
5259
const limit = options?.limit ?? DEFAULT_BATCH_SIZE;
53-
const now = new Date().toISOString();
60+
const now = new Date();
61+
const reclaimed = await reclaimStaleRunningJobs(supabase, now);
5462

5563
const { data: jobs } = await supabase
5664
.from("agent_jobs")
5765
.select("*")
5866
.eq("status", "pending")
59-
.lte("scheduled_at", now)
67+
.lte("scheduled_at", now.toISOString())
6068
.order("scheduled_at", { ascending: true })
6169
.limit(limit);
6270

@@ -69,7 +77,35 @@ export async function processAgentJobs(options?: {
6977
else failed += 1;
7078
}
7179

72-
return { processed: (jobs ?? []).length, succeeded, failed };
80+
return {
81+
processed: (jobs ?? []).length,
82+
succeeded,
83+
failed,
84+
reclaimed,
85+
};
86+
}
87+
88+
/** Requeue jobs stuck mid-run (deploy kill, OOM, timeout). */
89+
export async function reclaimStaleRunningJobs(
90+
supabase: ReturnType<typeof createAdminClient> = createAdminClient(),
91+
now = new Date(),
92+
staleMs = STALE_RUNNING_MS,
93+
): Promise<number> {
94+
const cutoff = new Date(now.getTime() - staleMs).toISOString();
95+
const { data, error } = await supabase
96+
.from("agent_jobs")
97+
.update({
98+
status: "pending",
99+
started_at: null,
100+
scheduled_at: now.toISOString(),
101+
error_message: "reclaimed_stale_running",
102+
})
103+
.eq("status", "running")
104+
.lt("started_at", cutoff)
105+
.select("id");
106+
107+
if (error) return 0;
108+
return data?.length ?? 0;
73109
}
74110

75111
async function processOneJob(job: AgentJobRow): Promise<boolean> {

‎supabase/migrations/022_mark_sent_from_prep_statuses.sql‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ BEGIN
8080
AND e.level = CASE (p_payload_json->>'required_proof')
8181
WHEN 'L1' THEN 'L1'::public.escalation_level
8282
WHEN 'L2' THEN 'L2'::public.escalation_level
83+
WHEN 'L3' THEN 'L3'::public.escalation_level
8384
ELSE 'L1'::public.escalation_level
8485
END
8586
AND e.status IN ('sent', 'response_received', 'timeout')
@@ -136,4 +137,5 @@ BEGIN
136137
END;
137138
$$;
138139

139-
COMMENT ON FUNCTION public.transition_case IS 'Guarded state machine;
140+
COMMENT ON FUNCTION public.transition_case IS
141+
'Guarded state machine; mark_sent allowed from prep statuses (022).';

‎supabase/migrations/023_prior_proof_l3.sql‎

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
-- 023: prior-level proof CASE must map L3 (L4 mark-sent).
2+
3+
CREATE OR REPLACE FUNCTION public.transition_case(
4+
p_case_id UUID,
5+
p_to_status public.case_status,
6+
p_trigger TEXT,
7+
p_actor_type public.actor_type DEFAULT 'system',
8+
p_actor_id TEXT DEFAULT NULL,
9+
p_payload_json JSONB DEFAULT '{}'
10+
)
11+
RETURNS public.cases
12+
LANGUAGE plpgsql
13+
SECURITY DEFINER
14+
SET search_path = public
15+
AS $$
16+
DECLARE
17+
v_case public.cases;
18+
v_from public.case_status;
19+
v_allowed BOOLEAN := FALSE;
20+
BEGIN
21+
-- Authorization guard (added in 018). service_role is trusted (route-level auth);
22+
-- everyone else must hold editor access or be an operator.
23+
IF COALESCE(auth.jwt() ->> 'role', '') <> 'service_role'
24+
AND NOT public.is_operator()
25+
AND NOT public.has_case_access(p_case_id, 'editor') THEN
26+
RAISE EXCEPTION 'forbidden'
27+
USING ERRCODE = '42501',
28+
HINT = '{"error":"forbidden","guard":"case_access"}';
29+
END IF;
30+
31+
SELECT * INTO v_case FROM public.cases WHERE id = p_case_id FOR UPDATE;
32+
IF NOT FOUND THEN
33+
RAISE EXCEPTION 'case_not_found' USING ERRCODE = 'P0002';
34+
END IF;
35+
36+
v_from := v_case.status;
37+
38+
IF v_from = p_to_status THEN
39+
RETURN v_case; -- idempotent no-op
40+
END IF;
41+
42+
-- Valid transitions per BUILD_SPEC §4 state machine
43+
v_allowed := CASE
44+
WHEN v_from = 'new' AND p_to_status = 'intake_scoping' AND p_trigger = 'evidence.submitted' THEN TRUE
45+
WHEN v_from = 'intake_scoping' AND p_to_status = 'monitoring' AND p_trigger = 'intake.classified' THEN TRUE
46+
WHEN v_from = 'monitoring' AND p_to_status = 'evidence_building' AND p_trigger = 'checklist.complete' THEN TRUE
47+
WHEN v_from = 'monitoring' AND p_to_status = 'closed' AND p_trigger IN ('user.abandon', 'inactive_30d') THEN TRUE
48+
WHEN v_from = 'evidence_building' AND p_to_status = 'escalation' AND p_trigger = 'bundle.ready' THEN TRUE
49+
WHEN v_from IN ('new', 'intake_scoping', 'monitoring', 'evidence_building', 'escalation', 'retried')
50+
AND p_to_status = 'awaiting_response'
51+
AND p_trigger = 'user.mark_sent' THEN TRUE
52+
WHEN v_from = 'awaiting_response' AND p_to_status = 'verified' AND p_trigger IN ('response.received', 'user.confirm_unfreeze') THEN TRUE
53+
WHEN v_from = 'awaiting_response' AND p_to_status = 'escalation' AND p_trigger = 'response.timeout' THEN TRUE
54+
WHEN v_from = 'awaiting_response' AND p_to_status = 'stalled' AND p_trigger = 'inactive_45d' THEN TRUE
55+
WHEN v_from = 'verified' AND p_to_status = 'resolved' AND p_trigger = 'resolution.confirmed' THEN TRUE
56+
WHEN v_from = 'resolved' AND p_to_status = 'public_pressure' AND p_trigger = 'user.opt_in_stats' THEN TRUE
57+
WHEN v_from = 'resolved' AND p_to_status = 'closed' AND p_trigger = 'bundle.delivered' THEN TRUE
58+
WHEN v_from = 'stalled' AND p_to_status = 'retried' AND p_trigger = 'user.reopen' THEN TRUE
59+
WHEN v_from = 'retried' AND p_to_status = 'escalation' AND p_trigger = 'new.strategy' THEN TRUE
60+
WHEN v_from = 'escalation' AND p_to_status = 'human_escalation'
61+
AND p_trigger IN ('low_confidence', 'cost_cap', 'user.request') THEN TRUE
62+
WHEN v_from = 'human_escalation' AND p_to_status = 'closed' AND p_trigger = 'ops.handoff' THEN TRUE
63+
-- Operator overrides
64+
WHEN public.is_operator() AND p_trigger LIKE 'ops.%' THEN TRUE
65+
ELSE FALSE
66+
END;
67+
68+
IF NOT v_allowed THEN
69+
RAISE EXCEPTION 'invalid_transition: % -> % via %', v_from, p_to_status, p_trigger
70+
USING ERRCODE = 'P0001',
71+
HINT = jsonb_build_object('error', 'guard_failed', 'from', v_from, 'to', p_to_status, 'trigger', p_trigger)::TEXT;
72+
END IF;
73+
74+
-- Guard: mark_sent requires prior level proof for L2+
75+
IF p_trigger = 'user.mark_sent' AND (p_payload_json->>'escalation_level') IN ('L2', 'L3', 'L4') THEN
76+
IF NOT EXISTS (
77+
SELECT 1 FROM public.escalations e
78+
WHERE e.case_id = p_case_id
79+
AND e.level = CASE (p_payload_json->>'required_proof')
80+
WHEN 'L1' THEN 'L1'::public.escalation_level
81+
WHEN 'L2' THEN 'L2'::public.escalation_level
82+
WHEN 'L3' THEN 'L3'::public.escalation_level
83+
ELSE 'L1'::public.escalation_level
84+
END
85+
AND e.status IN ('sent', 'response_received', 'timeout')
86+
) THEN
87+
RAISE EXCEPTION 'guard_failed: has_prior_level_proof'
88+
USING ERRCODE = 'P0001',
89+
HINT = '{"error":"guard_failed","guard":"has_prior_level_proof"}';
90+
END IF;
91+
END IF;
92+
93+
-- Guard: resolved requires confirmation
94+
IF p_to_status = 'resolved' THEN
95+
IF NOT (
96+
(p_payload_json->>'resolution_confirmed_by') IS NOT NULL
97+
OR EXISTS (
98+
SELECT 1 FROM public.evidence ev
99+
WHERE ev.case_id = p_case_id
100+
AND ev.evidence_type = 'bank_release_letter'
101+
AND ev.deleted_at IS NULL
102+
)
103+
) THEN
104+
RAISE EXCEPTION 'guard_failed: resolution_proof_required'
105+
USING ERRCODE = 'P0001';
106+
END IF;
107+
END IF;
108+
109+
UPDATE public.cases SET
110+
status = p_to_status,
111+
last_activity_at = now(),
112+
resolved_at = CASE WHEN p_to_status = 'resolved' THEN now() ELSE resolved_at END,
113+
closed_at = CASE WHEN p_to_status = 'closed' THEN now() ELSE closed_at END,
114+
stalled_at = CASE WHEN p_to_status = 'stalled' THEN now() ELSE stalled_at END,
115+
resolution_type = COALESCE((p_payload_json->>'resolution_type')::public.resolution_type, resolution_type),
116+
resolution_confirmed_by = COALESCE(p_payload_json->>'resolution_confirmed_by', resolution_confirmed_by),
117+
released_amount_paise = COALESCE((p_payload_json->>'released_amount_paise')::BIGINT, released_amount_paise)
118+
WHERE id = p_case_id
119+
RETURNING * INTO v_case;
120+
121+
INSERT INTO public.action_logs (case_id, actor_type, actor_id, action, payload_json)
122+
VALUES (
123+
p_case_id, p_actor_type, p_actor_id,
124+
'transition.' || p_to_status,
125+
jsonb_build_object('from', v_from, 'to', p_to_status, 'trigger', p_trigger) || COALESCE(p_payload_json, '{}')
126+
);
127+
128+
PERFORM public.append_swarm_event(
129+
p_case_id, 'MONITOR', 'status_transition',
130+
format('Case moved from %s to %s', v_from, p_to_status),
131+
'info', NULL,
132+
jsonb_build_object('from', v_from, 'to', p_to_status, 'trigger', p_trigger)
133+
);
134+
135+
RETURN v_case;
136+
END;
137+
$$;
138+
139+
COMMENT ON FUNCTION public.transition_case IS
140+
'Guarded state machine; mark_sent prep statuses + L3 prior proof (023).';
141+

0 commit comments

Comments
 (0)