diff --git a/docs/adr/0008-modelling-runs-filters-to-distributor.md b/docs/adr/0008-modelling-runs-filters-to-distributor.md index 0fb7785b..bbe0985c 100644 --- a/docs/adr/0008-modelling-runs-filters-to-distributor.md +++ b/docs/adr/0008-modelling-runs-filters-to-distributor.md @@ -26,17 +26,22 @@ explicit marker when trigger wiring lands. This is that wiring. - **A Modelling run is recorded as an app-created task** — the BulkUpload trigger convention (`triggerAddressMatching`): Next.js inserts the `tasks` row (service `modelling_run`, `source_id` = portfolio, status `waiting`) - plus one config `sub_task` whose `inputs` JSON carries - `{scenario_ids, filters, previewed_property_count, triggered_by}`, then - passes `task_id` to the distributor, which attaches its execution subtasks - to the same task. **No new table, no migration.** Run history and the - per-scenario in-flight badge read the portfolio's own `modelling_run` tasks - (portfolio-scoped, a handful of rows) and parse the app-authored inputs - JSON; execution progress ("61 of 104 ready") falls out of the existing - subtask-count summary pattern. -- **The distributor receives `{task_id, portfolio_id, scenario_ids, filters}`** - — filters, never property-id lists. The payload stays tiny at any portfolio - size; the config subtask is the durable copy of the same request. + whose **`inputs` column** (added for this — mirrors `sub_task.inputs`) + carries `{portfolio_id, scenario_ids, filters, previewed_property_count, + triggered_by}` as JSON, then passes `task_id` to the distributor, which + attaches its execution sub-tasks to the same task. **One new column, no new + table.** Run history and the per-scenario in-flight badge read the + portfolio's own `modelling_run` tasks (portfolio-scoped, a handful of rows) + and parse the app-authored inputs JSON; execution progress falls out of the + existing subtask-count summary pattern. Sub-task granularity is the + distributor's **batches**, so progress is shown as a percentage — batch + counts are internally consistent but never comparable to property or plan + numbers. +- **The distributor (`POST /v1/modelling/trigger-run`) receives + `{task_id, portfolio_id, scenario_ids, filters}`** — filters, never + property-id lists. The payload stays tiny at any portfolio size; the task's + `inputs` is the durable copy of the same request (plus provenance the + distributor doesn't need). - **Dispatch failure marks the task `failed`** (an improvement on the bulk-upload flow, which leaves failed triggers at `waiting`), so history never shows a ghost run as pending. @@ -63,7 +68,7 @@ explicit marker when trigger wiring lands. This is that wiring. run and the execution share one record by construction. Scenario badges gain a third derived state — "Modelling in progress" — and a per-portfolio run history lists each run's filters, counts, initiator and derived status - with subtask-count progress. + with batch-based percentage progress. - **Concurrent runs warn, never block**: triggering a scenario with a run in flight shows who started it and when, but proceeds. Plans append and latest-per-property wins on read, so overlap is safe; a hard lock could leak diff --git a/src/app/db/schema/tasks/tasks.ts b/src/app/db/schema/tasks/tasks.ts index badce0d0..e778d0d7 100644 --- a/src/app/db/schema/tasks/tasks.ts +++ b/src/app/db/schema/tasks/tasks.ts @@ -18,6 +18,11 @@ export const tasks = pgTable("tasks", { sourceId: text("source_id"), // identifier for the source (e.g., portfolio_id value) + // The task's request inputs, mirroring sub_task.inputs (JSON as text; could + // later change to JSONB). For a modelling run this is the run's config — + // portfolio id, scenario ids, filters (ADR-0008). + inputs: text("inputs"), + updatedAt: timestamp("updated_at", { precision: 6, withTimezone: true }) .defaultNow() .$onUpdate(() => new Date()) diff --git a/src/app/portfolio/[slug]/modelling/run/RunModellingClient.tsx b/src/app/portfolio/[slug]/modelling/run/RunModellingClient.tsx index 34202925..68b6c691 100644 --- a/src/app/portfolio/[slug]/modelling/run/RunModellingClient.tsx +++ b/src/app/portfolio/[slug]/modelling/run/RunModellingClient.tsx @@ -434,21 +434,7 @@ export default function RunModellingClient({ {r.previewedPropertyCount.toLocaleString()} - {r.status.state === "in_progress" ? ( - <> - - {r.status.ready.toLocaleString()} of {r.status.total.toLocaleString()} ready - - - - - - ) : ( - r.planCount.toLocaleString() - )} + {r.planCount.toLocaleString()} {STATUS_LABELS[r.status.state]} + {/* Work is chunked into batches, so show a percentage — + the counts never line up with property/plan numbers */} + {r.status.state === "in_progress" && ( + <> + + {Math.round((100 * r.status.ready) / Math.max(1, r.status.total))}% done + + + + + + )} ))} diff --git a/src/lib/modellingRuns/model.test.ts b/src/lib/modellingRuns/model.test.ts index 656a72fd..3767388f 100644 --- a/src/lib/modellingRuns/model.test.ts +++ b/src/lib/modellingRuns/model.test.ts @@ -44,7 +44,7 @@ describe("normaliseRunFilters", () => { }); describe("buildRunRecord", () => { - it("produces the app-authored task row and config-subtask inputs (ADR-0008)", () => { + it("produces the app-authored task row carrying the run config as its inputs (ADR-0008)", () => { const record = buildRunRecord({ portfolioId: "814", scenarioIds: ["12", "34"], @@ -58,27 +58,33 @@ describe("buildRunRecord", () => { source: "portfolio_id", sourceId: "814", status: "waiting", + inputs: JSON.stringify({ + portfolio_id: 814, + scenario_ids: [12, 34], + filters: { postcodes: ["B93 8SU"], property_types: ["House"] }, + previewed_property_count: 214, + triggered_by: "khalim@domna.homes", + }), }); - expect(record.configInputs).toEqual({ + // What the distributor receives — provenance fields stay on the task only + expect(record.dispatchPayload).toEqual({ portfolio_id: 814, scenario_ids: [12, 34], filters: { postcodes: ["B93 8SU"], property_types: ["House"] }, - previewed_property_count: 214, - triggered_by: "khalim@domna.homes", }); }); }); describe("parseRunConfig", () => { it("round-trips a stored run record back to the app's shape", () => { - const { configInputs } = buildRunRecord({ + const { task } = buildRunRecord({ portfolioId: "814", scenarioIds: ["12"], filters: { builtForms: ["Detached", "Unknown"] }, previewedPropertyCount: 57, triggeredBy: "priya@domna.homes", }); - expect(parseRunConfig(JSON.stringify(configInputs))).toEqual({ + expect(parseRunConfig(task.inputs)).toEqual({ scenarioIds: ["12"], filters: { builtForms: ["Detached", "Unknown"] }, previewedPropertyCount: 57, diff --git a/src/lib/modellingRuns/model.ts b/src/lib/modellingRuns/model.ts index c00fe73a..973bf2b5 100644 --- a/src/lib/modellingRuns/model.ts +++ b/src/lib/modellingRuns/model.ts @@ -168,9 +168,9 @@ export interface RunConfigInputs { } /** - * The rows a Modelling run persists (ADR-0008): an app-authored task (the - * BulkUpload trigger convention) plus the config-subtask inputs that make the - * run's request durable and history-readable. + * The row a Modelling run persists (ADR-0008): an app-authored task whose + * `inputs` carries the run's config — the durable, history-readable record. + * Sub-tasks are the distributor's: per-property execution work. */ export function buildRunRecord(req: RunRequest): { task: { @@ -179,13 +179,22 @@ export function buildRunRecord(req: RunRequest): { source: "portfolio_id"; sourceId: string; status: "waiting"; + inputs: string; }; - configInputs: RunConfigInputs; + /** What the distributor receives — provenance fields stay on the task. */ + dispatchPayload: Omit; } { const filters: RunConfigInputs["filters"] = {}; if (req.filters.postcodes) filters.postcodes = req.filters.postcodes; if (req.filters.propertyTypes) filters.property_types = req.filters.propertyTypes; if (req.filters.builtForms) filters.built_forms = req.filters.builtForms; + const configInputs: RunConfigInputs = { + portfolio_id: Number(req.portfolioId), + scenario_ids: req.scenarioIds.map(Number), + filters, + previewed_property_count: req.previewedPropertyCount, + triggered_by: req.triggeredBy, + }; return { task: { taskSource: `Modelling run – ${req.scenarioIds.length} scenario${req.scenarioIds.length === 1 ? "" : "s"}`, @@ -193,13 +202,12 @@ export function buildRunRecord(req: RunRequest): { source: "portfolio_id", sourceId: req.portfolioId, status: "waiting", + inputs: JSON.stringify(configInputs), }, - configInputs: { - portfolio_id: Number(req.portfolioId), - scenario_ids: req.scenarioIds.map(Number), + dispatchPayload: { + portfolio_id: configInputs.portfolio_id, + scenario_ids: configInputs.scenario_ids, filters, - previewed_property_count: req.previewedPropertyCount, - triggered_by: req.triggeredBy, }, }; } diff --git a/src/lib/modellingRuns/server.ts b/src/lib/modellingRuns/server.ts index d5ef3984..63f03c31 100644 --- a/src/lib/modellingRuns/server.ts +++ b/src/lib/modellingRuns/server.ts @@ -13,13 +13,11 @@ import { } from "./model"; import { builtFormTypeSql, propertyTypeSql } from "@/lib/services/epcSources"; -// The config subtask that makes a run's request durable (ADR-0008); execution -// subtasks attached by the distributor carry other service values. -export const RUN_CONFIG_SERVICE = "modelling_run_config"; - // Distributor entry point on the Model backend. It fans the run out to -// workers, attaching execution subtasks to the task_id we pass (ADR-0008). -const DISTRIBUTOR_ENDPOINT = "/v1/plan/trigger-run"; +// workers, attaching execution sub-tasks to the task_id we pass (ADR-0008). +// NOTE: sub-task granularity is BATCHES, not properties — progress counts are +// internally consistent but never comparable to the previewed property count. +const DISTRIBUTOR_ENDPOINT = "/v1/modelling/trigger-run"; type BackendResult = { ok: true; data: T } | { ok: false; status: number; message: string }; @@ -62,8 +60,9 @@ export type TriggerRunOutcome = | { kind: "dispatch_failed"; status: number; message: string }; /** - * Record + dispatch a Modelling run: app-authored task and config subtask - * (BulkUpload trigger convention), then hand task_id to the distributor. A + * Record + dispatch a Modelling run: one app-authored task whose `inputs` + * carries the run's config (BulkUpload trigger convention), then hand + * task_id to the distributor — sub_tasks are its, one per property. A * failed dispatch marks the task failed so history never shows a ghost run * as pending (ADR-0008). */ @@ -94,23 +93,12 @@ export async function triggerModellingRun(args: { .insert(tasks) .values({ ...record.task, jobStarted: now }) .returning(); - await db.insert(subTasks).values({ - taskId: task.id, - status: "waiting", - service: RUN_CONFIG_SERVICE, - inputs: JSON.stringify(record.configInputs), - }); // Minimal dispatch payload — previewed count and triggered_by are app - // provenance, already durable on the config subtask keyed by this task_id. + // provenance, already durable on the task's inputs. const dispatch = await callBackend( DISTRIBUTOR_ENDPOINT, - { - task_id: task.id, - portfolio_id: record.configInputs.portfolio_id, - scenario_ids: record.configInputs.scenario_ids, - filters: record.configInputs.filters, - }, + { task_id: task.id, ...record.dispatchPayload }, args.sessionToken, ); if (!dispatch.ok) { @@ -210,9 +198,10 @@ export interface RunHistoryRow { } /** - * Run history for a portfolio: its modelling_run tasks with app-authored - * config parsed back out and progress from execution-subtask counts. One - * grouped query — portfolio-scoped, never per-run probes. + * Run history for a portfolio: its modelling_run tasks with the app-authored + * config parsed from task inputs and progress from sub-task counts (one + * sub-task per property, distributor-attached). One grouped query — + * portfolio-scoped, never per-run probes. */ export async function listModellingRuns( portfolioId: string, @@ -224,10 +213,10 @@ export async function listModellingRuns( jobStarted: tasks.jobStarted, jobCompleted: tasks.jobCompleted, status: tasks.status, - configInputs: sql`max(${subTasks.inputs}) filter (where ${subTasks.service} = ${RUN_CONFIG_SERVICE})`, - total: sql`count(${subTasks.id}) filter (where ${subTasks.service} is distinct from ${RUN_CONFIG_SERVICE})::int`, - completed: sql`count(case when lower(${subTasks.status}) in ('completed','complete') and ${subTasks.service} is distinct from ${RUN_CONFIG_SERVICE} then 1 end)::int`, - failed: sql`count(case when lower(${subTasks.status}) in ('failed','failure','error') and ${subTasks.service} is distinct from ${RUN_CONFIG_SERVICE} then 1 end)::int`, + configInputs: tasks.inputs, + total: sql`count(${subTasks.id})::int`, + completed: sql`count(case when lower(${subTasks.status}) in ('completed','complete') then 1 end)::int`, + failed: sql`count(case when lower(${subTasks.status}) in ('failed','failure','error') then 1 end)::int`, }) .from(tasks) .leftJoin(subTasks, eq(subTasks.taskId, tasks.id))