Workflow reconciliation reports

Monitoring finds known failures. Reconciliation detects missing work and mismatches that produced no FastHook error. Compare a bounded set of expected business events with Workflow Activity, then verify successful FastHook runs against the provider or system of record.

Choose a stable join key

| Key | Use | Limitation | | --- | --- | --- | | FastHook request_id | Exact comparison when the producer or ingest log records it | External systems may not store it automatically. | | Provider event ID | Provider-to-FastHook ingress reconciliation | It may require request detail or provider logs rather than the Activity summary. | | Business identifier | Order, invoice, ticket, commit, or record reconciliation | A field mapping error can change or omit it. | | Provider result object ID | Verify the final Action side effect | Available only after the provider accepted or created the object. |

Do not reconcile by timestamps alone. Clock differences, retries, polling windows, and delayed provider delivery make time an interval filter, not a unique join key.

Prepare expected input

Create expected.jsonl with one JSON object per expected FastHook request:

JSON
{"request_id":"req_order_1001","business_key":"order_1001"}
{"request_id":"req_order_1002","business_key":"order_1002"}
{"request_id":"req_order_1003","business_key":"order_1003"}

The file must not contain payloads or credentials. Export it from the producer, ingest ledger, or another trusted system of record.

FastHook-side reconciliation script

Save this as fasthook-reconcile.mjs and run it with Node.js 18 or later:

JS
import { readFile } from "node:fs/promises";

const API_BASE = (process.env.FASTHOOK_API_BASE || "https://api.fasthook.io").replace(/\/$/, "");
const API_KEY = process.env.FASTHOOK_API_KEY;
const TEAM_ID = process.env.FASTHOOK_TEAM_ID;
const SOURCE_ID = process.env.FASTHOOK_SOURCE_ID;
const WINDOW_FROM = process.env.FASTHOOK_WINDOW_FROM;
const WINDOW_TO = process.env.FASTHOOK_WINDOW_TO || "now";
const INPUT_FILE = process.argv[2] || "expected.jsonl";

if (!API_KEY || !TEAM_ID || !SOURCE_ID || !WINDOW_FROM) {
  console.error("FASTHOOK_API_KEY, FASTHOOK_TEAM_ID, FASTHOOK_SOURCE_ID, and FASTHOOK_WINDOW_FROM are required");
  process.exit(3);
}

function parseJsonLines(text) {
  return text.split(/\r?\n/).filter(Boolean).map((line, index) => {
    const row = JSON.parse(line);
    if (!row.request_id || typeof row.request_id !== "string") {
      throw new Error(`Line ${index + 1} has no string request_id`);
    }
    return row;
  });
}

async function main() {
  const expected = parseJsonLines(await readFile(INPUT_FILE, "utf8"));
  const duplicateExpectedIds = expected.map((row) => row.request_id)
    .filter((id, index, all) => all.indexOf(id) !== index);
  if (duplicateExpectedIds.length) {
    throw new Error(`Expected input contains duplicate request IDs: ${[...new Set(duplicateExpectedIds)].join(", ")}`);
  }

  const query = new URLSearchParams({
    source_id: SOURCE_ID,
    from: WINDOW_FROM,
    to: WINDOW_TO,
    limit: "100"
  });
  const response = await fetch(`${API_BASE}/v1/workflow-activity?${query}`, {
    headers: {
      authorization: `Bearer ${API_KEY}`,
      "x-team-id": TEAM_ID,
      accept: "application/json"
    },
    signal: AbortSignal.timeout(30_000)
  });
  if (!response.ok) throw new Error(`Workflow Activity returned HTTP ${response.status}`);

  const activity = await response.json();
  const runs = Array.isArray(activity.models) ? activity.models : [];
  const total = Number(activity.total || 0);
  if (total > runs.length) {
    console.log(JSON.stringify({
      schema_version: 1,
      status: "incomplete",
      reason: "activity_window_overflow",
      returned: runs.length,
      total,
      action: "Split the time range and rerun; do not use this report for reconciliation."
    }, null, 2));
    process.exitCode = 3;
    return;
  }

  const runsByRequest = new Map();
  for (const run of runs) {
    const list = runsByRequest.get(run.request_id) || [];
    list.push(run);
    runsByRequest.set(run.request_id, list);
  }

  const expectedIds = new Set(expected.map((row) => row.request_id));
  const rows = expected.map((row) => {
    const matches = runsByRequest.get(row.request_id) || [];
    const statuses = [...new Set(matches.map((run) => run.status))].sort();
    let outcome = "missing";
    if (statuses.includes("failed")) outcome = "failed";
    else if (statuses.includes("queued") || statuses.includes("running")) outcome = "in_progress";
    else if (statuses.length === 1 && statuses[0] === "succeeded") outcome = "succeeded";
    else if (statuses.length) outcome = "mixed";
    return {
      request_id: row.request_id,
      business_key: row.business_key || null,
      outcome,
      run_count: matches.length,
      statuses,
      run_ids: matches.map((run) => run.id),
      workflow_ids: [...new Set(matches.map((run) => run.workflow_id))],
      error_codes: [...new Set(matches.map((run) => run.error_code).filter(Boolean))]
    };
  });

  const unexpected = runs.filter((run) => !expectedIds.has(run.request_id)).map((run) => ({
    request_id: run.request_id,
    run_id: run.id,
    workflow_id: run.workflow_id,
    status: run.status,
    error_code: run.error_code || null
  }));
  const counts = rows.reduce((result, row) => {
    result[row.outcome] = (result[row.outcome] || 0) + 1;
    return result;
  }, {});
  const duplicateRuns = rows.filter((row) => row.run_count > 1);
  const status = rows.every((row) => row.outcome === "succeeded") && !unexpected.length && !duplicateRuns.length
    ? "matched" : "differences";

  console.log(JSON.stringify({
    schema_version: 1,
    generated_at: new Date().toISOString(),
    status,
    team_id: TEAM_ID,
    source_id: SOURCE_ID,
    window: { from: WINDOW_FROM, to: WINDOW_TO },
    expected_count: expected.length,
    activity_count: runs.length,
    counts,
    duplicate_run_request_ids: duplicateRuns.map((row) => row.request_id),
    unexpected,
    rows
  }, null, 2));
  process.exitCode = status === "matched" ? 0 : 2;
}

main().catch((error) => {
  console.error(JSON.stringify({
    schema_version: 1,
    status: "report_error",
    error: error instanceof Error ? error.message : "unknown_report_error"
  }));
  process.exitCode = 3;
});

The report classifies missing, failed, in_progress, succeeded, and mixed expected rows, plus duplicate run request IDs and unexpected Activity rows.

One request can intentionally start more than one active Workflow for the same Source. In that design, run_count > 1 is not automatically a duplicate side effect. Compare the observed workflow_ids with the expected Workflow set before escalating.

Split large windows

Workflow Activity returns at most 100 rows. If total > count, the example exits with an incomplete report. Split the interval into smaller, non-overlapping windows until every response is complete, then merge by run ID.

Do not simply add q=request_id for every expected row at high volume. Use bounded time windows and Source filters first, then retrieve individual run details only for differences.

Verify provider outcomes

A succeeded FastHook run means its executed Steps completed according to their adapters. Full business reconciliation must still compare the final provider result or system of record.

Extend each row without copying the full provider object:

JSON
{
  "provider_check": {
    "status": "matched",
    "object_id": "invoice_123",
    "business_key": "order_1001",
    "checked_at": "2026-08-27T12:10:00.000Z"
  }
}

Use an authenticated provider API or provider export owned by your team. Never copy provider tokens into mappings or reconciliation output.

Recovery decisions

| Difference | Safe next step | | --- | --- | | Missing request and provider never sent it | Repair the producer/provider subscription before creating replacement work. | | Missing Workflow run but request exists | Check Workflow status, Source selection, and routing evidence. | | Failed run | Inspect the first failed Step and automatic retry history before manual recovery. | | In-progress run | Wait for the expected Delay/retry window or investigate a stale run. | | FastHook succeeded, provider object missing | Reconcile provider request logs and asynchronous processing before repeating the Action. | | More provider objects than expected | Stop repeats, identify idempotency failures, and approve compensation explicitly. |

Reconciliation is read-only evidence. Any replay, retry, compensation, credential rotation, or graph edit should follow the incident response runbook and change-control process.