David/WORKFLOW-APP
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: Task Kanban UI + Python task worker

Browse source 
Adds the visible front-end and the worker back-end for the agent
task system. The PocketBase schema, API routes, and ToastProvider
were already in the previous commit; this commit completes the
picture.

UI
- app/tasks/{page,KanbanBoard,TaskCard,CardModal,TaskDetail,api,columns}
  Full Kanban with six columns (Backlog, Ready, In Progress,
  Blocked, Done, Failed), drag-and-drop between columns, quick-add
  chips, type/priority filters, task detail panel with retry/cancel/
  delete, 5s polling, transition toasts.
- components/NavBar.tsx: 'Tasks' link in primary nav.
- app/bizportal-2/BizPortal2Explorer.tsx: 'Send to Kanban' button
  on the result view — dispatches the current search to the agent
  queue as a 'ready' task.

Back-end
- scripts/task_worker.py: PocketBase polling worker with two
  lanes. Deterministic handles bizportal_search/reserve, pdf_generate,
  email_send. Custom lane writes to /tmp/agent_inbox/ for the
  LLM agent to pick up. Atomic claim via PB filter, auto-retry up
  to 3x, structured log on each task.
- scripts/bizportal_service.py: bugfix — _is_logged_in was using
  a non-CSS selector so session-recovery was a no-op. Also made
  the login flow re-runnable mid-session (handles BizPortal's
  ~20-minute session timeout transparently).
- scripts/init_pb_schema.py: duration_ms no longer capped at 0.

Verified end-to-end:
- CK search via worker: 1 company, 3 directors, ~10s
- ID search via worker: 4 companies, 7 total directors, ~34s
- Auto-retry + status transitions + toast fires
This commit is contained in:
Dave 2026-08-17 18:57:02 +02:00
parent 2fb979b98a
commit c349998fa1
13 changed files with 1723 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
app/api/tasks/route.ts
View file

@ -56,7 +56,11 @@ export async function GET(req: NextRequest) {
const result = await fetchCollection<PBRecord>(COLLECTION, {
page: 1,
perPage: Math.min(limit, 500),
sort: "-priority,-created",
// PB on this instance doesn't support sort on the auto-managed
// `created` column. Use priority (desc) then id (desc) — id is
// a nanoid prefix so newer tasks come last when sorted desc, and
// the worker can poll deterministically.
sort: "-priority,-id",
filter: filters.length > 0 ? filters.join(" && ") : undefined,
});
return NextResponse.json(result);

52
app/bizportal-2/BizPortal2Explorer.tsx
View file

@ -167,6 +167,10 @@ export default function BizPortal2Explorer() {
<div className="space-y-2">
<p className="text-sm text-muted">No results found.</p>
{data.log && data.log.length > 0 && <LogBlock log={data.log} />}
<SendToKanbanButton
type={activeTab}
query={query}
/>
</div>
)}
</div>
@ -225,6 +229,53 @@ function LogBlock({ log }: { log: string[] }) {
);
}
function SendToKanbanButton({ type, query }: { type: SearchType; query: string }) {
const [sending, setSending] = useState(false);
const [done, setDone] = useState(false);
async function send() {
setSending(true);
try {
const res = await fetch("/api/tasks", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
title: `BizPortal ${type.toUpperCase()} search — ${query}`,
task_type: "bizportal_search",
payload: { type, query },
priority: "normal",
status: "ready",
created_by: "bizportal-2",
}),
});
if (!res.ok) throw new Error(`HTTP ${res.status}`);
setDone(true);
} catch (e) {
// best-effort
} finally {
setSending(false);
}
}
if (done) {
return (
<a
href="/tasks"
className="inline-flex items-center gap-2 text-xs font-semibold text-green-700 hover:underline"
>
✅ Task queued — open Kanban →
</a>
);
}
return (
<button
onClick={send}
disabled={sending}
className="inline-flex items-center gap-2 px-3 py-1.5 rounded-md border border-border bg-surface text-fg text-xs font-semibold hover:border-accent disabled:opacity-50"
>
{sending ? "Sending…" : "Send to Kanban →"}
</button>
);
}
// ── Result display ───────────────────────────────────────────────────────
interface DoneViewProps {
@ -485,6 +536,7 @@ function DoneView({ results, activeTab, query, durationMs, log }: DoneViewProps)
)}
</div>
<div className="flex items-center gap-2">
<SendToKanbanButton type={activeTab} query={query} />
{selectedCount > 0 ? (
<>
<span className="text-xs font-semibold text-fg">{selectedCount} selected</span>

402
app/tasks/CardModal.tsx Normal file
View file

@ -0,0 +1,402 @@
// app/tasks/CardModal.tsx
// New / edit card modal. Fields adapt to the selected task_type.
"use client";
import { useEffect, useState } from "react";
import {
TASK_TYPE_LABEL,
PRIORITY_LABEL,
type TaskType,
type TaskPriority,
type TaskStatus,
} from "./columns";
import type { Task, CreateTaskInput } from "./api";
interface CardModalProps {
open: boolean;
onClose: () => void;
onSubmit: (input: CreateTaskInput) => Promise<void>;
initial?: Partial<Task>;
mode?: "create" | "edit";
}
const TASK_TYPES: TaskType[] = [
"bizportal_search",
"bizportal_reserve",
"pdf_generate",
"email_send",
"custom",
];
const PRIORITIES: TaskPriority[] = ["low", "normal", "high", "urgent"];
const STATUSES: TaskStatus[] = ["backlog", "ready", "blocked"];
export default function CardModal({ open, onClose, onSubmit, initial, mode = "create" }: CardModalProps) {
const [title, setTitle] = useState("");
const [description, setDescription] = useState("");
const [taskType, setTaskType] = useState<TaskType>("bizportal_search");
const [priority, setPriority] = useState<TaskPriority>("normal");
const [status, setStatus] = useState<TaskStatus>("backlog");
const [assignee, setAssignee] = useState("agent");
const [submitting, setSubmitting] = useState(false);
const [error, setError] = useState<string | null>(null);
// Payload fields per task type
const [bpType, setBpType] = useState<"id" | "ck">("ck");
const [bpQuery, setBpQuery] = useState("");
const [bpNames, setBpNames] = useState<string>("");
const [bpDryRun, setBpDryRun] = useState(false);
const [pdfTemplate, setPdfTemplate] = useState("");
const [pdfSource, setPdfSource] = useState("");
const [emailTo, setEmailTo] = useState("");
const [emailSubject, setEmailSubject] = useState("");
const [emailBody, setEmailBody] = useState("");
const [customJson, setCustomJson] = useState("");
useEffect(() => {
if (!open) return;
setError(null);
if (initial) {
setTitle(initial.title ?? "");
setDescription(initial.description ?? "");
setTaskType(initial.task_type ?? "bizportal_search");
setPriority(initial.priority ?? "normal");
setStatus(initial.status ?? "backlog");
setAssignee(initial.assignee ?? "agent");
const p = (initial.payload ?? {}) as Record<string, any>;
setBpType(p.type === "id" ? "id" : "ck");
setBpQuery(p.query ?? "");
setBpNames((p.names ?? []).join("\n"));
setBpDryRun(!!p.dry_run);
setPdfTemplate(p.template ?? "");
setPdfSource(p.source ?? "");
setEmailTo(p.to ?? "");
setEmailSubject(p.subject ?? "");
setEmailBody(p.body ?? "");
setCustomJson(p.raw ? JSON.stringify(p.raw, null, 2) : "");
} else {
setTitle("");
setDescription("");
setTaskType("bizportal_search");
setPriority("normal");
setStatus("backlog");
setAssignee("agent");
setBpType("ck");
setBpQuery("");
setBpNames("");
setBpDryRun(false);
setPdfTemplate("");
setPdfSource("");
setEmailTo("");
setEmailSubject("");
setEmailBody("");
setCustomJson("");
}
}, [open, initial]);
if (!open) return null;
function buildPayload(): Record<string, unknown> {
switch (taskType) {
case "bizportal_search":
return { type: bpType, query: bpQuery.trim() };
case "bizportal_reserve":
return {
names: bpNames.split("\n").map((s) => s.trim()).filter(Boolean),
dry_run: bpDryRun,
};
case "pdf_generate":
return { template: pdfTemplate.trim(), source: pdfSource.trim() };
case "email_send":
return { to: emailTo.trim(), subject: emailSubject.trim(), body: emailBody };
case "custom":
try {
return { raw: customJson ? JSON.parse(customJson) : null };
} catch (e: any) {
throw new Error(`Custom JSON is invalid: ${e.message}`);
}
}
}
async function handleSubmit(e: React.FormEvent) {
e.preventDefault();
setError(null);
if (!title.trim()) {
setError("Title is required");
return;
}
setSubmitting(true);
try {
const payload = buildPayload();
await onSubmit({
title: title.trim(),
description: description.trim() || undefined,
task_type: taskType,
payload,
priority,
status,
assignee,
created_by: "user",
});
onClose();
} catch (err: any) {
setError(err.message ?? String(err));
} finally {
setSubmitting(false);
}
}
return (
<div
className="fixed inset-0 z-40 bg-black/40 flex items-start justify-center pt-12 px-4 overflow-y-auto"
onClick={onClose}
role="dialog"
aria-modal="true"
>
<div
className="bg-surface rounded-lg border border-border shadow-xl w-full max-w-lg"
onClick={(e) => e.stopPropagation()}
>
<form onSubmit={handleSubmit} className="p-5 space-y-4">
<div className="flex items-center justify-between">
<h2 className="text-lg font-bold text-fg">
{mode === "edit" ? "Edit task" : "New task"}
</h2>
<button
type="button"
onClick={onClose}
className="text-muted hover:text-fg text-sm"
aria-label="Close"
>
✕
</button>
</div>
<Field label="Title">
<input
type="text"
value={title}
onChange={(e) => setTitle(e.target.value)}
placeholder="e.g. Reserve name AMPLEFORTH PTY LTD"
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm focus:outline-none focus:ring-2 focus:ring-accent"
autoFocus
required
/>
</Field>
<Field label="Description">
<textarea
value={description}
onChange={(e) => setDescription(e.target.value)}
placeholder="Optional — context for the agent"
rows={2}
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm focus:outline-none focus:ring-2 focus:ring-accent"
/>
</Field>
<div className="grid grid-cols-2 gap-3">
<Field label="Type">
<select
value={taskType}
onChange={(e) => setTaskType(e.target.value as TaskType)}
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
>
{TASK_TYPES.map((t) => (
<option key={t} value={t}>
{TASK_TYPE_LABEL[t]}
</option>
))}
</select>
</Field>
<Field label="Priority">
<select
value={priority}
onChange={(e) => setPriority(e.target.value as TaskPriority)}
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
>
{PRIORITIES.map((p) => (
<option key={p} value={p}>
{PRIORITY_LABEL[p]}
</option>
))}
</select>
</Field>
</div>
<div className="grid grid-cols-2 gap-3">
<Field label="Status">
<select
value={status}
onChange={(e) => setStatus(e.target.value as TaskStatus)}
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
>
{STATUSES.map((s) => (
<option key={s} value={s}>
{s.replace("_", " ")}
</option>
))}
</select>
</Field>
<Field label="Assignee">
<input
type="text"
value={assignee}
onChange={(e) => setAssignee(e.target.value)}
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
/>
</Field>
</div>
<hr className="border-border" />
<h3 className="text-sm font-semibold text-fg">Payload</h3>
{taskType === "bizportal_search" ? (
<div className="grid grid-cols-3 gap-3">
<Field label="Search by">
<select
value={bpType}
onChange={(e) => setBpType(e.target.value as "id" | "ck")}
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
>
<option value="ck">CK No.</option>
<option value="id">Director ID</option>
</select>
</Field>
<div className="col-span-2">
<Field label="Query">
<input
type="text"
value={bpQuery}
onChange={(e) => setBpQuery(e.target.value)}
placeholder={bpType === "ck" ? "K2022808158" : "8709225206083"}
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
/>
</Field>
</div>
</div>
) : null}
{taskType === "bizportal_reserve" ? (
<div className="space-y-3">
<Field label="Names (one per line, in priority order)">
<textarea
value={bpNames}
onChange={(e) => setBpNames(e.target.value)}
rows={4}
placeholder={"AMPLEFORTH PTY LTD\nAMPLEFORTH TRADING\nAMPLEFORTH SA\nAMPLEFORTH HOLDINGS"}
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm font-mono"
/>
</Field>
<label className="flex items-center gap-2 text-sm text-fg">
<input
type="checkbox"
checked={bpDryRun}
onChange={(e) => setBpDryRun(e.target.checked)}
/>
Dry run (stop before payment — no R50 spent)
</label>
</div>
) : null}
{taskType === "pdf_generate" ? (
<div className="grid grid-cols-2 gap-3">
<Field label="Template">
<input
type="text"
value={pdfTemplate}
onChange={(e) => setPdfTemplate(e.target.value)}
placeholder="e.g. cor9_1_approval"
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
/>
</Field>
<Field label="Source">
<input
type="text"
value={pdfSource}
onChange={(e) => setPdfSource(e.target.value)}
placeholder="path or record id"
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
/>
</Field>
</div>
) : null}
{taskType === "email_send" ? (
<div className="space-y-3">
<Field label="To">
<input
type="email"
value={emailTo}
onChange={(e) => setEmailTo(e.target.value)}
placeholder="client@example.com"
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
/>
</Field>
<Field label="Subject">
<input
type="text"
value={emailSubject}
onChange={(e) => setEmailSubject(e.target.value)}
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
/>
</Field>
<Field label="Body">
<textarea
value={emailBody}
onChange={(e) => setEmailBody(e.target.value)}
rows={4}
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-sm"
/>
</Field>
</div>
) : null}
{taskType === "custom" ? (
<Field label="Custom payload (JSON)">
<textarea
value={customJson}
onChange={(e) => setCustomJson(e.target.value)}
rows={6}
placeholder='{ "instruction": "do this and that" }'
className="w-full px-3 py-2 rounded-md border border-border bg-bg text-fg text-xs font-mono"
/>
</Field>
) : null}
{error ? (
<div className="text-xs text-red-700 bg-red-50 border border-red-300 rounded px-3 py-2">
{error}
</div>
) : null}
<div className="flex justify-end gap-2 pt-2">
<button
type="button"
onClick={onClose}
className="px-4 py-2 rounded-md text-sm text-muted hover:text-fg"
>
Cancel
</button>
<button
type="submit"
disabled={submitting}
className="px-5 py-2 rounded-md bg-accent text-accent-on text-sm font-semibold hover:bg-accent-hover disabled:opacity-60"
>
{submitting ? "Saving…" : mode === "edit" ? "Save" : "Create task"}
</button>
</div>
</form>
</div>
</div>
);
}
function Field({ label, children }: { label: string; children: React.ReactNode }) {
return (
<label className="block">
<span className="text-xs font-semibold text-muted mb-1 block">{label}</span>
{children}
</label>
);
}

349
app/tasks/KanbanBoard.tsx Normal file
View file

@ -0,0 +1,349 @@
// app/tasks/KanbanBoard.tsx
// Client component — the full Kanban: columns, drag-and-drop, polling,
// toasts, modal/detail panels.
"use client";
import { useEffect, useMemo, useRef, useState } from "react";
import {
listTasks,
createTask,
patchTask,
deleteTask,
retryTask,
cancelTask,
type Task,
type CreateTaskInput,
} from "./api";
import { COLUMNS, DONE_COLLAPSE_DAYS, type TaskStatus, type TaskType, type TaskPriority } from "./columns";
import TaskCard from "./TaskCard";
import CardModal from "./CardModal";
import TaskDetail from "./TaskDetail";
import { useToast, maybeRequestNotificationPermission, fireBrowserNotification } from "@/components/Toast";
const POLL_MS = 5000;
export default function KanbanBoard() {
const [tasks, setTasks] = useState<Task[]>([]);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<string | null>(null);
const [modalOpen, setModalOpen] = useState(false);
const [modalInitial, setModalInitial] = useState<Partial<Task> | undefined>();
const [modalMode, setModalMode] = useState<"create" | "edit">("create");
const [detailTask, setDetailTask] = useState<Task | null>(null);
const [draggingId, setDraggingId] = useState<string | null>(null);
const [dragOverCol, setDragOverCol] = useState<TaskStatus | null>(null);
const [includeDone, setIncludeDone] = useState(false);
const [filterType, setFilterType] = useState<TaskType | "">("");
const [filterPriority, setFilterPriority] = useState<TaskPriority | "">("");
const { toast } = useToast();
const lastStatusesRef = useRef<Map<string, TaskStatus>>(new Map());
async function fetchTasks() {
try {
const items = await listTasks({ include_done: true, limit: 500 });
// Detect transitions for toast fires
const prev = lastStatusesRef.current;
const next = new Map<string, TaskStatus>();
for (const t of items) {
next.set(t.id, t.status);
const old = prev.get(t.id);
if (old && old !== t.status) {
if (t.status === "done") {
toast({ kind: "success", title: "Task done", body: t.title });
fireBrowserNotification("Task done", { body: t.title });
} else if (t.status === "failed") {
toast({ kind: "error", title: "Task failed", body: t.error ?? t.title });
fireBrowserNotification("Task failed", { body: t.title });
} else if (t.status === "in_progress" && old === "ready") {
toast({ kind: "info", title: "Agent started", body: t.title, duration: 3000 });
}
}
}
lastStatusesRef.current = next;
setTasks(items);
setError(null);
} catch (err: any) {
setError(err.message);
} finally {
setLoading(false);
}
}
useEffect(() => {
fetchTasks();
maybeRequestNotificationPermission();
const t = setInterval(fetchTasks, POLL_MS);
return () => clearInterval(t);
// eslint-disable-next-line react-hooks/exhaustive-deps
}, []);
// ── Filtering ─────────────────────────────────────────────
const filtered = useMemo(() => {
return tasks.filter((t) => {
if (filterType && t.task_type !== filterType) return false;
if (filterPriority && t.priority !== filterPriority) return false;
if (!includeDone && (t.status === "done" || t.status === "failed")) {
// still show recent ones (within 1 day) for context
const completed = t.completed_at
? new Date(t.completed_at.includes("T") ? t.completed_at : t.completed_at.replace(" ", "T") + "Z").getTime()
: 0;
if (Date.now() - completed > 24 * 60 * 60 * 1000) return false;
}
return true;
});
}, [tasks, filterType, filterPriority, includeDone]);
const byColumn = useMemo(() => {
const out: Record<TaskStatus, Task[]> = {
backlog: [],
ready: [],
in_progress: [],
blocked: [],
done: [],
failed: [],
};
for (const t of filtered) out[t.status].push(t);
return out;
}, [filtered]);
// ── Handlers ──────────────────────────────────────────────
async function handleCreate(input: CreateTaskInput) {
const created = await createTask(input);
toast({ kind: "success", title: "Task created", body: created.title });
await fetchTasks();
}
async function handleEdit(input: CreateTaskInput) {
if (!modalInitial?.id) return;
await patchTask(modalInitial.id, {
title: input.title,
description: input.description,
priority: input.priority,
payload: input.payload,
assignee: input.assignee,
});
toast({ kind: "success", title: "Task updated" });
await fetchTasks();
}
function openNewCard(prefill?: Partial<Task>) {
setModalInitial(prefill);
setModalMode("create");
setModalOpen(true);
}
function openEditCard(task: Task) {
setModalInitial(task);
setModalMode("edit");
setModalOpen(true);
setDetailTask(null);
}
function quickAdd(type: TaskType) {
openNewCard({ task_type: type, status: "ready", priority: "normal" });
}
// ── Drag-and-drop ─────────────────────────────────────────
function handleDragStart(e: React.DragEvent, id: string) {
setDraggingId(id);
e.dataTransfer.effectAllowed = "move";
e.dataTransfer.setData("text/plain", id);
}
function handleDragEnd() {
setDraggingId(null);
setDragOverCol(null);
}
function handleDragOverCol(e: React.DragEvent, col: TaskStatus) {
e.preventDefault();
e.dataTransfer.dropEffect = "move";
setDragOverCol(col);
}
function handleDragLeaveCol() {
setDragOverCol(null);
}
async function handleDrop(e: React.DragEvent, target: TaskStatus) {
e.preventDefault();
setDragOverCol(null);
const id = e.dataTransfer.getData("text/plain");
if (!id) return;
const task = tasks.find((t) => t.id === id);
if (!task || task.status === target) {
setDraggingId(null);
return;
}
// The PATCH endpoint enforces valid transitions. If invalid, the
// API returns 409 — surface that to the user.
try {
await patchTask(id, { status: target });
toast({ kind: "info", title: `Moved to ${target.replace("_", " ")}`, body: task.title, duration: 2000 });
await fetchTasks();
} catch (err: any) {
toast({
kind: "warn",
title: "Move blocked",
body: err.message ?? "Invalid state transition",
});
} finally {
setDraggingId(null);
}
}
// ── Quick add chips ───────────────────────────────────────
const quickAddChips: { type: TaskType; label: string; icon: string }[] = [
{ type: "bizportal_search", label: "BizPortal search", icon: "🔍" },
{ type: "bizportal_reserve", label: "Name reservation", icon: "📝" },
{ type: "pdf_generate", label: "PDF", icon: "📄" },
{ type: "email_send", label: "Email", icon: "✉️" },
{ type: "custom", label: "Custom", icon: "🧩" },
];
return (
<div className="container-x py-6 space-y-4">
{/* ── Toolbar ────────────────────────────────────────── */}
<div className="flex flex-wrap items-center justify-between gap-3">
<div className="flex flex-wrap items-center gap-2">
<button
onClick={() => openNewCard()}
data-od-id="task-new"
className="px-4 py-2 rounded-md bg-accent text-accent-on text-sm font-semibold hover:bg-accent-hover"
>
+ New task
</button>
<span className="text-xs text-muted">or quick-add:</span>
{quickAddChips.map((c) => (
<button
key={c.type}
onClick={() => quickAdd(c.type)}
className="px-3 py-1.5 rounded-md border border-border bg-surface text-fg text-xs font-semibold hover:border-accent"
title={`Create ${c.label} task (pre-filled to Ready)`}
>
{c.icon} {c.label}
</button>
))}
</div>
<div className="flex items-center gap-2">
<select
value={filterType}
onChange={(e) => setFilterType(e.target.value as TaskType | "")}
className="px-2 py-1.5 rounded-md border border-border bg-surface text-fg text-xs"
>
<option value="">All types</option>
<option value="bizportal_search">BizPortal search</option>
<option value="bizportal_reserve">BizPortal reserve</option>
<option value="pdf_generate">PDF</option>
<option value="email_send">Email</option>
<option value="custom">Custom</option>
</select>
<select
value={filterPriority}
onChange={(e) => setFilterPriority(e.target.value as TaskPriority | "")}
className="px-2 py-1.5 rounded-md border border-border bg-surface text-fg text-xs"
>
<option value="">All priorities</option>
<option value="urgent">Urgent</option>
<option value="high">High</option>
<option value="normal">Normal</option>
<option value="low">Low</option>
</select>
<label className="flex items-center gap-1.5 text-xs text-muted">
<input
type="checkbox"
checked={includeDone}
onChange={(e) => setIncludeDone(e.target.checked)}
/>
Show done/failed
</label>
</div>
</div>
{error ? (
<div className="rounded-md border border-red-300 bg-red-50 text-red-800 px-4 py-2 text-sm">
Failed to load tasks: {error}
</div>
) : null}
{/* ── Columns ───────────────────────────────────────── */}
<div className="overflow-x-auto pb-2">
<div className="flex gap-3 min-w-max">
{COLUMNS.map((col) => {
const colTasks = byColumn[col.id] ?? [];
const isOver = dragOverCol === col.id;
const isDoneCol = col.id === "done";
const isFailedCol = col.id === "failed";
const showCollapsed = (isDoneCol || isFailedCol) && colTasks.length > 0;
return (
<div
key={col.id}
onDragOver={(e) => handleDragOverCol(e, col.id)}
onDragLeave={handleDragLeaveCol}
onDrop={(e) => handleDrop(e, col.id)}
data-od-id={`column-${col.id}`}
className={`w-72 shrink-0 flex flex-col rounded-md border-2 ${col.accent} ${
isOver ? "bg-accent/5" : "bg-surface-warm/50"
} transition-colors`}
>
<div className="px-3 py-2 border-b border-border flex items-center justify-between">
<div>
<h3 className="text-sm font-bold text-fg">{col.title}</h3>
<p className="text-[10px] text-muted">{col.hint}</p>
</div>
<span className="text-xs font-semibold px-2 py-0.5 rounded-full bg-surface border border-border">
{colTasks.length}
</span>
</div>
<div className="flex-1 p-2 space-y-2 min-h-[200px] max-h-[calc(100vh-260px)] overflow-y-auto">
{colTasks.length === 0 ? (
<p className="text-[11px] text-muted italic text-center py-4">
{loading ? "Loading…" : "Drop cards here"}
</p>
) : (
colTasks.map((task) => (
<TaskCard
key={task.id}
task={task}
isDragging={draggingId === task.id}
onClick={() => setDetailTask(task)}
onDragStart={(e) => handleDragStart(e, task.id)}
onDragEnd={handleDragEnd}
/>
))
)}
</div>
</div>
);
})}
</div>
</div>
<p className="text-[11px] text-muted text-center">
Auto-refreshes every {POLL_MS / 1000}s · {tasks.length} task{tasks.length === 1 ? "" : "s"} loaded
{DONE_COLLAPSE_DAYS ? ` · Done column auto-collapses after ${DONE_COLLAPSE_DAYS} days` : ""}
</p>
{/* ── Modals ────────────────────────────────────────── */}
<CardModal
open={modalOpen}
onClose={() => setModalOpen(false)}
onSubmit={modalMode === "edit" ? handleEdit : handleCreate}
initial={modalInitial}
mode={modalMode}
/>
{detailTask ? (
<TaskDetail
task={detailTask}
onClose={() => setDetailTask(null)}
onMutated={fetchTasks}
onToast={(kind, title, body) => toast({ kind, title, body })}
/>
) : null}
</div>
);
}

99
app/tasks/TaskCard.tsx Normal file
View file

@ -0,0 +1,99 @@
// app/tasks/TaskCard.tsx
// Compact card for a single task. Used inside the Kanban columns.
"use client";
import type { Task } from "./api";
import {
TASK_TYPE_META,
PRIORITY_COLORS,
PRIORITY_LABEL,
DONE_COLLAPSE_DAYS,
} from "./columns";
interface TaskCardProps {
task: Task;
onClick?: () => void;
onDragStart?: (e: React.DragEvent) => void;
onDragEnd?: (e: React.DragEvent) => void;
isDragging?: boolean;
}
export default function TaskCard({
task,
onClick,
onDragStart,
onDragEnd,
isDragging,
}: TaskCardProps) {
const meta = TASK_TYPE_META[task.task_type];
const priorityClass = PRIORITY_COLORS[task.priority];
const isDone = task.status === "done";
const isOld = isDone && isOlderThanDays(task.completed_at, DONE_COLLAPSE_DAYS);
return (
<div
draggable
onDragStart={onDragStart}
onDragEnd={onDragEnd}
onClick={onClick}
data-od-id={`task-card-${task.id}`}
className={`group rounded-md border border-border bg-surface p-3 cursor-grab active:cursor-grabbing transition-all ${
isDragging ? "opacity-50 scale-95" : "hover:border-accent hover:shadow-sm"
} ${isOld ? "opacity-60" : ""}`}
>
<div className="flex items-start justify-between gap-2 mb-2">
<h4 className="text-sm font-semibold text-fg leading-tight line-clamp-2">{task.title}</h4>
</div>
<div className="flex flex-wrap gap-1.5 mb-2">
<span className={`text-[10px] font-semibold uppercase tracking-wider px-1.5 py-0.5 rounded border ${meta.tone}`}>
{meta.icon} {meta.label}
</span>
<span className={`text-[10px] font-semibold uppercase tracking-wider px-1.5 py-0.5 rounded border ${priorityClass}`}>
{PRIORITY_LABEL[task.priority]}
</span>
{task.retry_count > 0 ? (
<span className="text-[10px] font-semibold uppercase tracking-wider px-1.5 py-0.5 rounded border bg-orange-50 text-orange-800 border-orange-300">
retry ×{task.retry_count}
</span>
) : null}
</div>
<div className="flex items-center justify-between text-[11px] text-muted">
<span>{relativeTime(task.created)}</span>
{task.duration_ms != null ? (
<span className="font-mono">{(task.duration_ms / 1000).toFixed(1)}s</span>
) : null}
</div>
{task.error ? (
<p className="mt-2 text-[11px] text-red-700 line-clamp-2" title={task.error}>
❌ {task.error}
</p>
) : null}
</div>
);
}
function relativeTime(iso: string | null): string {
if (!iso) return "—";
const t = new Date(iso.includes("T") ? iso : iso.replace(" ", "T") + "Z").getTime();
if (isNaN(t)) return "—";
const diff = Date.now() - t;
const sec = Math.floor(diff / 1000);
if (sec < 60) return `${sec}s ago`;
const min = Math.floor(sec / 60);
if (min < 60) return `${min}m ago`;
const hr = Math.floor(min / 60);
if (hr < 24) return `${hr}h ago`;
const day = Math.floor(hr / 24);
return `${day}d ago`;
}
function isOlderThanDays(iso: string | null, days: number): boolean {
if (!iso) return false;
const t = new Date(iso.includes("T") ? iso : iso.replace(" ", "T") + "Z").getTime();
if (isNaN(t)) return false;
return Date.now() - t > days * 24 * 60 * 60 * 1000;
}

183
app/tasks/TaskDetail.tsx Normal file
View file

@ -0,0 +1,183 @@
// app/tasks/TaskDetail.tsx
// Expanded panel for a single task. Shows log, result, error. Has
// retry, cancel, and delete buttons.
"use client";
import { useState } from "react";
import { TASK_TYPE_META, STATUS_LABEL, PRIORITY_LABEL } from "./columns";
import { type Task, retryTask, cancelTask, deleteTask } from "./api";
interface TaskDetailProps {
task: Task;
onClose: () => void;
onMutated: () => void;
onToast: (kind: "info" | "success" | "error" | "warn", title: string, body?: string) => void;
}
export default function TaskDetail({ task, onClose, onMutated, onToast }: TaskDetailProps) {
const [busy, setBusy] = useState<string | null>(null);
const meta = TASK_TYPE_META[task.task_type];
async function act(label: string, fn: () => Promise<any>) {
setBusy(label);
try {
await fn();
onMutated();
onToast("success", `${label} done`);
} catch (err: any) {
onToast("error", `${label} failed`, err.message);
} finally {
setBusy(null);
}
}
return (
<div
className="fixed inset-0 z-40 bg-black/40 flex items-start justify-center pt-12 px-4 overflow-y-auto"
onClick={onClose}
role="dialog"
aria-modal="true"
>
<div
className="bg-surface rounded-lg border border-border shadow-xl w-full max-w-2xl"
onClick={(e) => e.stopPropagation()}
>
<div className="p-5 space-y-4">
<div className="flex items-start justify-between gap-3">
<div className="flex-1 min-w-0">
<div className="flex flex-wrap items-center gap-2 mb-2">
<span className={`text-[10px] font-semibold uppercase tracking-wider px-1.5 py-0.5 rounded border ${meta.tone}`}>
{meta.icon} {meta.label}
</span>
<span className="text-[10px] font-semibold uppercase tracking-wider px-1.5 py-0.5 rounded border bg-blue-50 text-blue-700 border-blue-200">
{STATUS_LABEL[task.status]}
</span>
<span className="text-[10px] font-semibold uppercase tracking-wider px-1.5 py-0.5 rounded border bg-amber-50 text-amber-800 border-amber-200">
{PRIORITY_LABEL[task.priority]}
</span>
{task.retry_count > 0 ? (
<span className="text-[10px] font-semibold uppercase tracking-wider px-1.5 py-0.5 rounded border bg-orange-50 text-orange-800 border-orange-300">
retry ×{task.retry_count}
</span>
) : null}
</div>
<h2 className="text-xl font-bold text-fg leading-tight">{task.title}</h2>
{task.description ? (
<p className="text-sm text-muted mt-2 whitespace-pre-wrap">{task.description}</p>
) : null}
</div>
<button
type="button"
onClick={onClose}
className="text-muted hover:text-fg text-sm"
aria-label="Close"
>
✕
</button>
</div>
<div className="grid grid-cols-2 gap-3 text-xs">
<Meta label="Assignee" value={task.assignee ?? "—"} />
<Meta label="Claimed by" value={task.claimed_by ?? "—"} />
<Meta label="Created" value={formatTime(task.created)} />
<Meta label="Claimed at" value={formatTime(task.claimed_at)} />
<Meta label="Started" value={formatTime(task.started_at)} />
<Meta label="Completed" value={formatTime(task.completed_at)} />
<Meta label="Duration" value={task.duration_ms != null ? `${(task.duration_ms / 1000).toFixed(2)}s` : "—"} />
<Meta label="Created by" value={task.created_by ?? "—"} />
</div>
{task.payload && Object.keys(task.payload as object).length > 0 ? (
<Section title="Payload">
<pre className="text-xs font-mono bg-bg p-3 rounded border border-border overflow-x-auto">
{JSON.stringify(task.payload, null, 2)}
</pre>
</Section>
) : null}
{task.error ? (
<Section title="Error">
<p className="text-xs text-red-700 bg-red-50 border border-red-300 p-3 rounded">
{task.error}
</p>
</Section>
) : null}
{task.log && task.log.length > 0 ? (
<Section title={`Log (${task.log.length} entries)`}>
<pre className="text-[11px] font-mono bg-bg p-3 rounded border border-border overflow-x-auto whitespace-pre-wrap max-h-60 overflow-y-auto">
{task.log.join("\n")}
</pre>
</Section>
) : null}
{task.result ? (
<Section title="Result">
<pre className="text-xs font-mono bg-bg p-3 rounded border border-border overflow-x-auto max-h-60 overflow-y-auto">
{JSON.stringify(task.result, null, 2)}
</pre>
</Section>
) : null}
<div className="flex flex-wrap gap-2 pt-2 border-t border-border">
{(task.status === "failed" || task.status === "done") && (
<button
onClick={() => act("Retry", () => retryTask(task.id))}
disabled={busy !== null}
className="px-3 py-1.5 rounded-md bg-blue-600 text-white text-xs font-semibold hover:bg-blue-700 disabled:opacity-50"
>
{busy === "Retry" ? "Retrying…" : "↻ Retry"}
</button>
)}
{task.status === "in_progress" && (
<button
onClick={() => act("Cancel", () => cancelTask(task.id))}
disabled={busy !== null}
className="px-3 py-1.5 rounded-md bg-amber-500 text-white text-xs font-semibold hover:bg-amber-600 disabled:opacity-50"
>
{busy === "Cancel" ? "Cancelling…" : "✕ Cancel"}
</button>
)}
<button
onClick={() => {
if (confirm(`Delete "${task.title}"? This cannot be undone.`)) {
act("Delete", () => deleteTask(task.id).then(onClose));
}
}}
disabled={busy !== null}
className="px-3 py-1.5 rounded-md border border-red-300 text-red-700 text-xs font-semibold hover:bg-red-50 disabled:opacity-50"
>
{busy === "Delete" ? "Deleting…" : "🗑 Delete"}
</button>
</div>
</div>
</div>
</div>
);
}
function Meta({ label, value }: { label: string; value: string }) {
return (
<div>
<p className="text-[10px] font-semibold text-muted uppercase tracking-wider mb-0.5">{label}</p>
<p className="text-fg">{value}</p>
</div>
);
}
function Section({ title, children }: { title: string; children: React.ReactNode }) {
return (
<div>
<h3 className="text-xs font-semibold text-muted uppercase tracking-wider mb-1.5">{title}</h3>
{children}
</div>
);
}
function formatTime(iso: string | null): string {
if (!iso) return "—";
const t = new Date(iso.includes("T") ? iso : iso.replace(" ", "T") + "Z");
if (isNaN(t.getTime())) return "—";
return t.toLocaleString();
}

122
app/tasks/api.ts Normal file
View file

@ -0,0 +1,122 @@
// app/tasks/api.ts
// Browser-safe client for the /api/tasks routes.
import type { TaskStatus, TaskPriority, TaskType } from "./columns";
export interface Task {
id: string;
collectionId: string;
collectionName: string;
created: string;
updated: string;
title: string;
description: string | null;
task_type: TaskType;
payload: Record<string, unknown> | null;
status: TaskStatus;
priority: TaskPriority;
assignee: string | null;
claimed_by: string | null;
claimed_at: string | null;
started_at: string | null;
completed_at: string | null;
duration_ms: number | null;
result: unknown;
error: string | null;
log: string[] | null;
retry_count: number;
created_by: string | null;
}
export interface ListParams {
status?: TaskStatus | "all";
task_type?: TaskType;
priority?: TaskPriority;
include_done?: boolean;
limit?: number;
}
export async function listTasks(params: ListParams = {}): Promise<Task[]> {
const qs = new URLSearchParams();
if (params.status) qs.set("status", params.status);
if (params.task_type) qs.set("task_type", params.task_type);
if (params.priority) qs.set("priority", params.priority);
if (params.include_done) qs.set("include_done", "true");
if (params.limit) qs.set("limit", String(params.limit));
const res = await fetch(`/api/tasks?${qs.toString()}`, { cache: "no-store" });
if (!res.ok) throw new Error(`list failed (${res.status})`);
const data = await res.json();
return data.items as Task[];
}
export async function getTask(id: string): Promise<Task> {
const res = await fetch(`/api/tasks/${id}`, { cache: "no-store" });
if (!res.ok) throw new Error(`get failed (${res.status})`);
return res.json();
}
export interface CreateTaskInput {
title: string;
description?: string;
task_type: TaskType;
payload?: Record<string, unknown>;
status?: TaskStatus;
priority?: TaskPriority;
assignee?: string;
created_by?: string;
}
export async function createTask(input: CreateTaskInput): Promise<Task> {
const res = await fetch(`/api/tasks`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(input),
});
if (!res.ok) {
const err = await res.json().catch(() => ({}));
throw new Error(err.error ?? `create failed (${res.status})`);
}
return res.json();
}
export async function patchTask(
id: string,
fields: Partial<Pick<Task, "title" | "description" | "priority" | "payload" | "assignee" | "status">>,
): Promise<Task> {
const res = await fetch(`/api/tasks/${id}`, {
method: "PATCH",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(fields),
});
if (!res.ok) {
const err = await res.json().catch(() => ({}));
throw new Error(err.error ?? `patch failed (${res.status})`);
}
return res.json();
}
export async function deleteTask(id: string): Promise<void> {
const res = await fetch(`/api/tasks/${id}`, { method: "DELETE" });
if (!res.ok) throw new Error(`delete failed (${res.status})`);
}
export async function retryTask(id: string): Promise<Task> {
const res = await fetch(`/api/tasks/${id}/retry`, { method: "POST" });
if (!res.ok) {
const err = await res.json().catch(() => ({}));
throw new Error(err.error ?? `retry failed (${res.status})`);
}
const data = await res.json();
return data.task as Task;
}
export async function cancelTask(id: string): Promise<Task> {
const res = await fetch(`/api/tasks/${id}/cancel`, { method: "POST" });
if (!res.ok) {
const err = await res.json().catch(() => ({}));
throw new Error(err.error ?? `cancel failed (${res.status})`);
}
const data = await res.json();
return data.task as Task;
}

57
app/tasks/columns.ts Normal file
View file

@ -0,0 +1,57 @@
// app/tasks/columns.ts
// Column definitions for the Kanban. Status → display config.
export type TaskStatus = "backlog" | "ready" | "in_progress" | "blocked" | "done" | "failed";
export type TaskPriority = "low" | "normal" | "high" | "urgent";
export type TaskType = "bizportal_search" | "bizportal_reserve" | "pdf_generate" | "email_send" | "custom";
export const COLUMNS: { id: TaskStatus; title: string; hint: string; accent: string }[] = [
{ id: "backlog", title: "Backlog", hint: "Not yet ready", accent: "border-border" },
{ id: "ready", title: "Ready", hint: "Agent can pick up", accent: "border-blue-300" },
{ id: "in_progress", title: "In Progress", hint: "Agent is working", accent: "border-amber-300" },
{ id: "blocked", title: "Blocked", hint: "Needs human input", accent: "border-red-300" },
{ id: "done", title: "Done", hint: "Completed", accent: "border-green-300" },
{ id: "failed", title: "Failed", hint: "Error — retry or discard", accent: "border-red-400" },
];
export const PRIORITY_COLORS: Record<TaskPriority, string> = {
low: "bg-gray-100 text-gray-700 border-gray-200",
normal: "bg-blue-50 text-blue-700 border-blue-200",
high: "bg-amber-100 text-amber-800 border-amber-300",
urgent: "bg-red-100 text-red-800 border-red-300",
};
export const TASK_TYPE_META: Record<TaskType, { label: string; icon: string; tone: string }> = {
bizportal_search: { label: "BizPortal · search", icon: "🔍", tone: "bg-emerald-50 text-emerald-800 border-emerald-200" },
bizportal_reserve: { label: "BizPortal · reserve", icon: "📝", tone: "bg-emerald-50 text-emerald-800 border-emerald-200" },
pdf_generate: { label: "PDF · generate", icon: "📄", tone: "bg-violet-50 text-violet-800 border-violet-200" },
email_send: { label: "Email · send", icon: "✉️", tone: "bg-sky-50 text-sky-800 border-sky-200" },
custom: { label: "Custom", icon: "🧩", tone: "bg-slate-50 text-slate-800 border-slate-200" },
};
export const TASK_TYPE_LABEL: Record<TaskType, string> = {
bizportal_search: "BizPortal search",
bizportal_reserve: "BizPortal name reservation",
pdf_generate: "PDF generation",
email_send: "Email send",
custom: "Custom task",
};
export const PRIORITY_LABEL: Record<TaskPriority, string> = {
low: "Low",
normal: "Normal",
high: "High",
urgent: "Urgent",
};
export const STATUS_LABEL: Record<TaskStatus, string> = {
backlog: "Backlog",
ready: "Ready",
in_progress: "In Progress",
blocked: "Blocked",
done: "Done",
failed: "Failed",
};
// 7 days in done auto-collapse
export const DONE_COLLAPSE_DAYS = 7;

23
app/tasks/page.tsx Normal file
View file

@ -0,0 +1,23 @@
// app/tasks/page.tsx
// Server component. Renders the Kanban board.
import { Suspense } from "react";
import PageHeader from "@/components/PageHeader";
import KanbanBoard from "./KanbanBoard";
export const dynamic = "force-dynamic";
export default function TasksPage() {
return (
<>
<PageHeader
eyebrow="Agent tasks"
title="Task Kanban"
subtitle="Assign work to the agent. Move a card to Ready to dispatch it. The deterministic Python worker handles BizPortal tasks; the LLM lane handles Custom tasks."
/>
<Suspense fallback={<div className="container-x py-10 text-muted">Loading board…</div>}>
<KanbanBoard />
</Suspense>
</>
);
}

1
components/NavBar.tsx
View file

@ -25,6 +25,7 @@ const LINKS = [
{ href: "/", label: "Overview", icon: IconHome },
{ href: "/bizportal", label: "BizPortal", icon: IconSearch },
{ href: "/bizportal-2", label: "BizPortal 2.0", icon: IconSearch, badge: "test" },
{ href: "/tasks", label: "Tasks", icon: IconClipboard },
{ href: "/directors", label: "Directors", icon: IconUsers },
{ href: "/companies", label: "Companies", icon: IconBuilding },
{ href: "/groups", label: "Groups", icon: IconTag },

52
scripts/bizportal_service.py
View file

@ -203,24 +203,31 @@ async def _save_storage_state() -> None:
async def _is_logged_in(page: Page) -> bool:
try:
return await page.locator(LOGIN_ID).count() == 0
return await page.locator(f"#{LOGIN_ID}").count() == 0
except Exception:
return False
async def _ensure_logged_in(page: Page, log_list: List[str]) -> None:
"""Reuse stored session if valid; otherwise log in once."""
"""Reuse stored session if valid; otherwise log in once.
Idempotent and recoverable: if the page is currently on the
login screen (session expired), this re-logs in. Caches the
fresh storage state at the end.
"""
# Always start by navigating to LOGIN_URL so we know the page
# state. If we're already logged in, the server redirects us to
# the dashboard and the login form won't be present.
log.info("Ensuring logged in (navigating to login URL)…")
log_list.append("Navigating to login URL to check session state")
await page.goto(LOGIN_URL, wait_until="domcontentloaded", timeout=NAV_TIMEOUT_MS)
await page.wait_for_timeout(1500)
if await _is_logged_in(page):
log_list.append("Already logged in (cached session)")
return
log_list.append("Login form present — submitting credentials")
log.info("Logging in…")
log_list.append("Navigating to login")
await page.goto(LOGIN_URL, wait_until="domcontentloaded", timeout=NAV_TIMEOUT_MS)
if await _is_logged_in(page):
log_list.append("Already logged in (after navigation)")
return
id_no = os.getenv("BIZPORTAL_ID_NO")
password = os.getenv("BIZPORTAL_PASSWORD")
if not id_no or not password:
@ -240,6 +247,18 @@ async def _ensure_logged_in(page: Page, log_list: List[str]) -> None:
log_list.append("Logged in and cached session")
log.info("Logged in")
async def _ensure_on_bizportal(page: Page, log_list: List[str]) -> None:
"""If a page lands on login mid-flow, re-login. Used by the
search and reserve endpoints after navigating to a BizPortal
page — handles expired sessions transparently."""
try:
if await page.locator(f"#{LOGIN_ID}").count() > 0:
log_list.append("Session expired — re-logging in")
await _ensure_logged_in(page, log_list)
except Exception:
pass
async def _parse_balance(page: Page) -> Optional[float]:
await page.goto(PROFILE_URL, wait_until="domcontentloaded", timeout=NAV_TIMEOUT_MS)
await page.wait_for_timeout(1500)
@ -453,9 +472,22 @@ async def search(req: SearchRequest):
log_list.append(f"GET {BIZPROFILE_URL}")
await page.goto(BIZPROFILE_URL, wait_until="domcontentloaded", timeout=NAV_TIMEOUT_MS)
await page.wait_for_timeout(1500)
# Recover from expired sessions — the page may have
# redirected to login. Re-login and try again.
if await page.locator(f"#{LOGIN_ID}").count() > 0:
log_list.append("Session expired — re-logging in")
await _ensure_logged_in(page, log_list)
await page.goto(BIZPROFILE_URL, wait_until="domcontentloaded", timeout=NAV_TIMEOUT_MS)
await page.wait_for_timeout(1500)
log_list.append(f"Select dropdown: {option}")
await page.locator(f"#{BIZPROFILE_DROPDOWN}").select_option(option)
try:
await page.locator(f"#{BIZPROFILE_DROPDOWN}").select_option(option, timeout=15000)
except PWTimeout:
raise RuntimeError(
f"Could not find search dropdown #{BIZPROFILE_DROPDOWN} on {page.url}. "
"Session may have expired mid-flow."
)
# Dropdown change triggers a postback — wait for the page to settle.
try:
await page.wait_for_load_state("domcontentloaded", timeout=15000)

2
scripts/init_pb_schema.py
View file

@ -60,7 +60,7 @@ TASKS_SCHEMA = {
{"name": "claimed_at", "type": "date", "required": False, "max": "", "min": ""},
{"name": "started_at", "type": "date", "required": False, "max": "", "min": ""},
{"name": "completed_at", "type": "date", "required": False, "max": "", "min": ""},
{"name": "duration_ms", "type": "number", "required": False, "min": 0, "max": 0, "noDecimal": True},
{"name": "duration_ms", "type": "number", "required": False, "min": 0, "noDecimal": True},
# ── Output ────────────────────────────────────────────────
{"name": "result", "type": "json", "required": False, "maxSize": 65535},

387
scripts/task_worker.py Normal file
View file

@ -0,0 +1,387 @@
#!/usr/bin/env python3
"""
task_worker.py
==============
PocketBase task worker — pulls ready tasks and dispatches to the
right sidecar. Two lanes:
--lane deterministic
- bizportal_search → http://127.0.0.1:7001/search
- bizportal_reserve → http://127.0.0.1:7001/reserve
- pdf_generate → shell out to pdf-creator
- email_send → SMTP via the user's email config
(no LLM calls — runs fast, deterministic)
--lane custom
- custom → write a JSON file to /tmp/agent_inbox/ for
the LLM agent to pick up. The agent is
expected to PATCH the task back via the
/api/tasks/[id]/complete endpoint.
Both lanes share the same atomic claim protocol. Only one worker
ever holds a given task.
Usage:
python3 scripts/task_worker.py --lane deterministic
python3 scripts/task_worker.py --lane custom
"""
import argparse
import json
import logging
import os
import smtplib
import subprocess
import sys
import time
import urllib.request
import urllib.error
from email.mime.text import MIMEText
from pathlib import Path
from typing import Any, Dict, Optional
# ── Config ──────────────────────────────────────────────────────
ROOT = Path(__file__).resolve().parent.parent
ENV_FILE = ROOT / ".env.local"
DEFAULT_PB_URL = "https://pb.agent.apphub.co.za"
DEFAULT_TASKS_API = "http://127.0.0.1:6001" # the Next.js side
DEFAULT_BIZPORTAL_SERVICE = "http://127.0.0.1:7001"
DEFAULT_AGENT_INBOX = "/tmp/agent_inbox"
DEFAULT_PDF_CREATOR = "/Volumes/KIOXIA/.agents/skills/pdf-creator"
POLL_INTERVAL_S = 2.0
HTTP_TIMEOUT_S = 120
logging.basicConfig(
level=os.getenv("LOG_LEVEL", "INFO"),
format="%(asctime)s [%(levelname)s] %(name)s — %(message)s",
)
log = logging.getLogger(f"task-worker[{os.getpid()}]")
def load_env() -> Dict[str, str]:
env = dict(os.environ)
if ENV_FILE.exists():
for line in ENV_FILE.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
env[k.strip()] = v.strip().strip('"').strip("'")
return env
# ── HTTP helpers ────────────────────────────────────────────────
def http_json(url: str, method: str = "GET", headers: Optional[Dict] = None, body: Any = None):
data = None
if body is not None:
data = json.dumps(body).encode("utf-8")
h = dict(headers or {})
if data is not None and "Content-Type" not in h:
h["Content-Type"] = "application/json"
req = urllib.request.Request(url, data=data, headers=h, method=method)
try:
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT_S) as resp:
payload = resp.read().decode("utf-8")
return resp.status, json.loads(payload) if payload else None
except urllib.error.HTTPError as e:
body = e.read().decode("utf-8", errors="ignore")
try:
return e.code, json.loads(body)
except Exception:
return e.code, {"raw": body[:500]}
except Exception as e:
return 0, {"error": str(e)}
# ── Task lifecycle ─────────────────────────────────────────────
def list_ready(env: Dict[str, str], lane: str, limit: int = 1) -> list:
"""Fetch ready tasks. Filter by what the lane can handle."""
pb_url = env.get("PB_URL", DEFAULT_PB_URL)
# We go through the Next.js API so the auth flow is consistent.
api = env.get("TASKS_API", DEFAULT_TASKS_API)
qs = []
qs.append("status=ready")
if lane == "deterministic":
# Deterministic lane takes everything except custom
pass
elif lane == "custom":
# LLM lane takes only custom
qs.append("task_type=custom")
qs.append(f"limit={limit}")
url = f"{api}/api/tasks?{'&'.join(qs)}"
code, data = http_json(url, headers={"Authorization": env.get("PB_TOKEN", "")})
if code != 200:
log.warning(f"list ready failed: {code} {data}")
return []
if isinstance(data, dict) and "items" in data:
return data["items"]
return []
def claim_task(env: Dict[str, str], task_id: str) -> Optional[Dict]:
api = env.get("TASKS_API", DEFAULT_TASKS_API)
url = f"{api}/api/tasks/{task_id}/claim"
code, data = http_json(
url,
method="POST",
headers={"Authorization": env.get("PB_TOKEN", "")},
body={"claimed_by": f"task_worker[{os.getpid()}]:{lane_label(env)}"},
)
if code == 200 and isinstance(data, dict) and data.get("ok"):
return data.get("task")
return None
def complete_task(env: Dict[str, str], task_id: str, result: Any, log_lines: list, duration_ms: int) -> bool:
api = env.get("TASKS_API", DEFAULT_TASKS_API)
url = f"{api}/api/tasks/{task_id}/complete"
code, data = http_json(
url,
method="POST",
headers={"Authorization": env.get("PB_TOKEN", "")},
body={"result": result, "log": log_lines, "duration_ms": duration_ms},
)
return code == 200 and isinstance(data, dict) and data.get("ok")
def fail_task(env: Dict[str, str], task_id: str, error_msg: str, log_lines: list) -> bool:
api = env.get("TASKS_API", DEFAULT_TASKS_API)
url = f"{api}/api/tasks/{task_id}/fail"
code, data = http_json(
url,
method="POST",
headers={"Authorization": env.get("PB_TOKEN", "")},
body={"error": error_msg, "log": log_lines, "retry": True},
)
return code == 200 and isinstance(data, dict) and data.get("ok")
def lane_label(env: Dict[str, str]) -> str:
return env.get("WORKER_LANE", "deterministic")
# ── Handlers ────────────────────────────────────────────────────
def handle_bizportal_search(env: Dict[str, str], payload: Dict) -> tuple[Any, list]:
log = [f"bizportal_search payload={payload}"]
service = env.get("BIZPORTAL_SERVICE_URL", DEFAULT_BIZPORTAL_SERVICE)
url = f"{service}/search"
code, data = http_json(url, method="POST", body=payload)
log.append(f"service responded {code}")
if code != 200 or not isinstance(data, dict):
raise RuntimeError(f"bizportal service error: {code} {data}")
if not data.get("ok"):
raise RuntimeError(data.get("error") or "search failed")
log.append(f"parsed {len(data.get('results') or [])} companies")
return data, log
def handle_bizportal_reserve(env: Dict[str, str], payload: Dict) -> tuple[Any, list]:
log = [f"bizportal_reserve payload={payload}"]
service = env.get("BIZPORTAL_SERVICE_URL", DEFAULT_BIZPORTAL_SERVICE)
url = f"{service}/reserve"
code, data = http_json(url, method="POST", body=payload)
log.append(f"service responded {code}")
if code != 200 or not isinstance(data, dict):
raise RuntimeError(f"bizportal service error: {code} {data}")
if not data.get("ok"):
raise RuntimeError(data.get("error") or "reserve failed")
log.append(f"reservation status={data.get('status')}")
return data, log
def handle_pdf_generate(env: Dict[str, str], payload: Dict) -> tuple[Any, list]:
"""Shell out to the pdf-creator skill. The skill is a directory
that contains a runnable script — we invoke it with a JSON
payload via stdin and read the result file path from stdout."""
log = [f"pdf_generate payload={payload}"]
skill = env.get("PDF_CREATOR_PATH", DEFAULT_PDF_CREATOR)
if not Path(skill).exists():
raise RuntimeError(f"pdf-creator not found at {skill}")
# Look for an entry-point script
entry = None
for candidate in ("run.sh", "main.py", "pdf_creator.py", "index.js"):
p = Path(skill) / candidate
if p.exists():
entry = p
break
if entry is None:
raise RuntimeError(f"no entry-point script found in {skill}")
log.append(f"invoking {entry}")
proc = subprocess.run(
[str(entry)],
input=json.dumps(payload),
capture_output=True,
text=True,
timeout=300,
)
log.append(f"rc={proc.returncode}")
if proc.returncode != 0:
log.append(f"stderr={proc.stderr[:500]}")
raise RuntimeError(f"pdf-creator failed (rc={proc.returncode}): {proc.stderr[:200]}")
result = {"stdout": proc.stdout.strip(), "stderr": proc.stderr.strip()}
return result, log
def handle_email_send(env: Dict[str, str], payload: Dict) -> tuple[Any, list]:
log = [f"email_send payload keys={list(payload.keys())}"]
to = payload.get("to")
subject = payload.get("subject", "")
body = payload.get("body", "")
if not to or not isinstance(to, str):
raise RuntimeError("email_send: 'to' is required")
smtp_host = env.get("SMTP_HOST")
smtp_user = env.get("SMTP_USER")
smtp_pass = env.get("SMTP_PASSWORD") or env.get("SMTP_PASS")
smtp_from = env.get("SMTP_FROM", smtp_user or "noreply@example.com")
smtp_port = int(env.get("SMTP_PORT", "587"))
if not smtp_host:
raise RuntimeError("SMTP_HOST not configured in .env.local")
msg = MIMEText(body)
msg["Subject"] = subject
msg["From"] = smtp_from
msg["To"] = to
log.append(f"connecting to {smtp_host}:{smtp_port} as {smtp_user}")
with smtplib.SMTP(smtp_host, smtp_port, timeout=30) as smtp:
smtp.starttls()
if smtp_user and smtp_pass:
smtp.login(smtp_user, smtp_pass)
smtp.send_message(msg)
log.append("sent OK")
return {"to": to, "subject": subject, "sent": True}, log
def handle_custom(env: Dict[str, str], task: Dict) -> tuple[Any, list]:
"""LLM lane — write the task to the agent's inbox directory. The
agent is expected to PATCH the task back via /complete or /fail.
We return a placeholder so the worker can mark this batch as
"dispatched" but the actual completion comes from the agent."""
inbox = Path(env.get("AGENT_INBOX", DEFAULT_AGENT_INBOX))
inbox.mkdir(parents=True, exist_ok=True)
payload = {
"task_id": task["id"],
"title": task["title"],
"description": task.get("description"),
"payload": task.get("payload"),
"result_endpoint": f"{env.get('TASKS_API', DEFAULT_TASKS_API)}/api/tasks/{task['id']}/complete",
"fail_endpoint": f"{env.get('TASKS_API', DEFAULT_TASKS_API)}/api/tasks/{task['id']}/fail",
}
out = inbox / f"{task['id']}.json"
out.write_text(json.dumps(payload, indent=2))
log_lines = [f"wrote {out}"]
# Do NOT mark complete — the LLM agent will do that. The task
# stays in 'in_progress' until the agent calls back.
raise _InboxDispatched(out, log_lines)
class _InboxDispatched(Exception):
"""Special exception signalling the LLM agent has been notified
and the worker should NOT call /complete — the agent will."""
def __init__(self, out_path: Path, log_lines: list):
self.out_path = out_path
self.log_lines = log_lines
HANDLERS = {
"bizportal_search": handle_bizportal_search,
"bizportal_reserve": handle_bizportal_reserve,
"pdf_generate": handle_pdf_generate,
"email_send": handle_email_send,
"custom": handle_custom,
}
# The custom lane only takes `custom` task type; the deterministic
# lane takes everything except `custom`.
DETERMINISTIC_TYPES = {"bizportal_search", "bizportal_reserve", "pdf_generate", "email_send"}
# ── Main loop ──────────────────────────────────────────────────
def run_one(env: Dict[str, str], lane: str, task: Dict) -> None:
"""Claim and execute a single task. Returns nothing — all state
transitions go through the API."""
task_type = task["task_type"]
title = task["title"]
if lane == "deterministic" and task_type not in DETERMINISTIC_TYPES:
log.debug(f"skip {task['id']} (type={task_type} not in deterministic lane)")
return
if lane == "custom" and task_type != "custom":
log.debug(f"skip {task['id']} (type={task_type} not in custom lane)")
return
log.info(f"claim {task_type} {task['id'][:12]}… '{title}'")
claimed = claim_task(env, task["id"])
if not claimed:
log.info(f" → already claimed by someone else")
return
handler = HANDLERS.get(task_type)
if handler is None:
fail_task(env, task["id"], f"no handler for task_type={task_type}", [f"unknown type {task_type}"])
return
t0 = time.time()
log_lines: list = []
try:
payload = claimed.get("payload") or {}
result, handler_log = handler(env, payload)
log_lines.extend(handler_log)
duration_ms = int((time.time() - t0) * 1000)
complete_task(env, task["id"], result, log_lines, duration_ms)
log.info(f" → done in {duration_ms}ms")
except _InboxDispatched as d:
# LLM agent has been notified — log and move on
log_lines.extend(d.log_lines)
log.info(f" → dispatched to agent inbox, awaiting callback")
except Exception as e:
log_lines.append(f"ERROR: {e}")
duration_ms = int((time.time() - t0) * 1000)
log.error(f" → failed after {duration_ms}ms: {e}")
fail_task(env, task["id"], str(e)[:4900], log_lines)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--lane", choices=["deterministic", "custom", "all"], default="all",
help="Which lane to run. 'all' = both as one process.")
parser.add_argument("--once", action="store_true", help="Run one pass and exit (for testing).")
args = parser.parse_args()
env = load_env()
env["WORKER_LANE"] = args.lane
lanes = ["deterministic", "custom"] if args.lane == "all" else [args.lane]
log.info(f"worker pid={os.getpid()} lane={lanes} poll={POLL_INTERVAL_S}s")
log.info(f"PB_URL={env.get('PB_URL', DEFAULT_PB_URL)}")
log.info(f"TASKS_API={env.get('TASKS_API', DEFAULT_TASKS_API)}")
log.info(f"BIZPORTAL_SERVICE_URL={env.get('BIZPORTAL_SERVICE_URL', DEFAULT_BIZPORTAL_SERVICE)}")
while True:
any_claimed = False
for lane in lanes:
try:
ready = list_ready(env, lane, limit=1)
except Exception as e:
log.warning(f"list_ready error: {e}")
continue
for task in ready:
any_claimed = True
try:
run_one(env, lane, task)
except Exception as e:
log.exception(f"run_one crashed: {e}")
if args.once:
break
if not any_claimed:
time.sleep(POLL_INTERVAL_S)
if __name__ == "__main__":
main()