Flows and scheduled jobs

• A flow profile (fp_…) is a named set of variant keys from the catalogue, for example {"report.delivery": {"mode": "async", "delay_s": 10800}}. Create it with POST /v1/flow-profiles ({name, description?, variants}; names are [a-z0-9-], 1 to 64 characters).
• A profile is complete on its own.Every key it does not set takes the platform default, which is today's behaviour. Profiles are never merged or layered, so what a test names is exactly what it gets.
• Versions are immutable. PATCH /v1/flow-profiles/:id writes a new version with the complete variants (not a merge) and moves current_version. Old versions stay readable (GET /v1/flow-profiles/:id/versions) and usable.
• Pinning. A reference is fp_x (the current version, looked up on each request) or fp_x@3 (version 3, forever). Every profile response carries ref, the pinned reference to its current version: send that from tests, so an edit someone makes in the portal never changes a running test.
• fp_platformis a reserved, read-only id that means today's behaviour, whatever the workspace default is.
• Create is idempotent by name and body. The same name with the same description and variants returns the existing profile with 200; a different body is 409 flow_profile_name_taken. The example fixture names profiles <name>-<hash8> from a hash of the variants, so two branches with different variants never collide.
• Archive, never delete. DELETE /v1/flow-profiles/:id archives it and frees the name at once. Requests that still name it keep working for 24 hours with X-DM-Flow-Warning: archived, then get 400 invalid_flow_profile. Profiles unused for 30 days that are neither the workspace default nor assigned to a synthetic account are archived automatically.
• Unknown keys or values outside a key's schema are 400 invalid_flow_variant with details: [{field, constraints}], at save time, never at request time.

curl -s -X POST https://api.dispute.markets/v1/flow-profiles \
  -H "Authorization: Bearer $DM_API_KEY" -H "Content-Type: application/json" \
  -H "Idempotency-Key: profile-report-3h-01" \
  -d '{"name":"report-3h","variants":{"report.delivery":{"mode":"async","delay_s":10800}}}'

{"id":"fp_01J9Z3…","object":"flow_profile","name":"report-3h","current_version":1,
 "ref":"fp_01J9Z3…@1","variants":{"report.delivery":{"mode":"async","delay_s":10800}},
 "body_hash":"sha256:…","used_by":{"synthetic_accounts":0,"workspace_default":false}, …}

A request runs under a profile when it sends X-DM-Flow: fp_…@N (or fp_…, or fp_platform). In the browser, a test sets the dm_flow cookie on the portal host (or the owner presses Use in this browser in Settings › Flows) and the portal forwards it as X-DM-Flow, the same way it forwards dm_correlation_id.

The workspace default

PUT /v1/flow-profiles/default with {"profile_id": "fp_…@N"} (or null / fp_platformto clear it) sets what requests without a header, and synthetic accounts without a profile of their own, resolve to. It applies to resources created from then on; existing ones keep their stamp. Only the owner's signed-in session can set it (Settings › Flows, or the dev panel's Flow defaults card): no API key can hold flow.default.write. It exists for people clicking in the portal. Tests should not depend on it: the example fixture sends fp_platform on every call so a default someone sets never reaches CI.

Every catalogue key has a scope that says when it is read:

The profile a request resolves to, first match wins:

1. X-DM-Flow, when the caller may opt in (the owner, or an owner-minted key with flow.read). Profile source header.
2. For a request authenticated as a synthetic account: the profile the owner assigned to that account. Source account.
3. The workspace default. Source tenant_default.
4. fp_platform, today's behaviour. Source platform.

webhook.delivery is read from the request that triggered the event when there is one, else from the stamp of the resource that changed (a lifecycle transition, a finished report).

Snapshot, not reference

A resource that acts later (a report, a dispute from a generate job, an evidence upload, a held webhook delivery, a scheduled job) stores flow: {profile_id, version, source, variants} with the concrete values of its stamped keys, even for fp_platform. Later work reads the snapshot, never the live profile, so editing or archiving a profile, or changing the workspace default, never changes a resource that already exists. A resource created before flows existed behaves as fp_platform.

Profile A makes reports slow; profile B makes download links short-lived. A report created under A and downloaded under B gets both, each from the place its key's scope says:

POST /v1/reports            X-DM-Flow: fp_A@1
  A = {"report.delivery": {"mode": "async", "delay_s": 3600}}

201 X-DM-Flow-Applied: fp_A@1
{"id":"rpt_…","status":"queued","ready_at":"<now + 1 h>",
 "flow":{"profile_id":"fp_A","version":1,"source":"header",
         "variants":{"report.delivery":{"mode":"async","delay_s":3600},"report.outcome":{"mode":"ok"}}}, …}

POST /v1/scheduled-jobs/fast-forward   X-DM-Flow: fp_B@1
  {"target_id": "rpt_…", "wait": true}      → the build runs now; the report is done
GET  /v1/reports/rpt_…/download-url    X-DM-Flow: fp_B@1
  B = {"report.download_url": {"ttl_s": 60}}
200 {"url":"https://…","expires_at":"<now + 60 s>"}

• B does not set report.delivery, but that does not make the report synchronous: it is a stamped key, read from the report's snapshot.
• A does not set report.download_url, but the link lives 60 seconds, not the default 300: it is a request key, read from B.
• The audit trail says which is which. The report.build row carries flow: {profile_id: "fp_A@1", profile_source: "header", variants: {"report.delivery": {"value": {…}, "source": "stamped"}}}; the report.create_download_url row carries flow: {profile_id: "fp_B@1", …, variants: {"report.download_url": {"value": {"ttl_s": 60}, "source": "request"}}}.

Every key, rendered from the same data as GET /v1/flows/catalogue(which also returns each key's full JSON Schema). ? marks an optional field; unsetmeans the key does nothing unless a profile sets it. “Emits” lists every error code, audit action and webhook event the variant can produce, so a test can assert on exactly the failure it asked for.

Copy one into POST /v1/flow-profiles (or start from it in Settings › Flows › Start from recipe).

{
  "name": "slow-report",
  "variants": {
    "report.delivery": {
      "mode": "async",
      "delay_s": 3600,
      "build_s": 60
    }
  }
}

{
  "name": "flaky-challenge",
  "variants": {
    "dispute.action_failure": {
      "actions": [
        "challenge"
      ],
      "mode": "first_n_per_target",
      "n": 1,
      "error": "scheme_unavailable"
    }
  }
}

{
  "name": "lose-fast",
  "variants": {
    "lifecycle.windows": {
      "evidence_s": 60,
      "rfi_s": 60
    },
    "lifecycle.review": {
      "duration_s": 60
    },
    "lifecycle.ruling": {
      "mode": "always_lost"
    }
  }
}

{
  "name": "noisy-webhooks",
  "variants": {
    "lifecycle.webhooks": {
      "enabled": true
    }
  }
}

{
  "name": "slow-scan",
  "variants": {
    "evidence.processing": {
      "delay_s": 600,
      "outcome": "available"
    }
  }
}

A synthetic account cannot choose a flow (its header and cookie are ignored), but the owner can assign it one. Requests authenticated as that account then resolve through the account's profile → the workspace default → fp_platform, for request keys on every request and for stamped keys when it creates something. Owner sessions and API keys are unaffected: a key never inherits an account's profile.

• Set it with flow_profile on POST /v1/team/synthetic-accounts and change it with PATCH /v1/team/synthetic-accounts/:id(omit it to keep the assignment). It applies from the account's next request, on every server. An unknown or archived profile is 400 invalid_flow_profile.
• Owner only. Sending flow_profile (any value, null included) needs flow.write on top of team.write; without it the whole request is 403 missing_permission and nothing is written. API keys cannot manage synthetic accounts at all (401 api_key_not_allowed).
• Reads return flow_profile: {ref, profile_id, name, version, archived_at} (or null), only to callers holding flow.read; it is omitted for everyone else. The synthetic_account.create and synthetic_account.update audit rows record it in changes.flow_profile.
• Audit rows of requests the account made, and the snapshots of what it created, carry profile_source: "account" (header, account, tenant_default or, in snapshots, platform).
• Archiving a profile in use is allowed. Profile reads and the archive response carry used_by: {synthetic_accounts, workspace_default}. For 24 hours the accounts keep the archived profile (responses carry X-DM-Flow-Warning: archived); after that their assignment is ignored and they resolve through the workspace default, and a background job clears each assignment with a synthetic_account.flow_profile_released audit row (changes: {flow_profile: {from: 'fp_…', to: null}}, reason: profile_archived).

The testing pattern: one account per worker

Give each parallel browser worker its own synthetic account and assign each account its profile once, in worker setup, so workers run different variants at the same time, request keys such as session.max_age and dispute.action_failure included. Never change an assignment inside a test: concurrent CI runs that share an account pool share its assignments. The example does this with a worker option (see the E2E guide):

import { expect, hasOwner, uiTest as test } from '../../src/dm';
import { syntheticAccounts } from '../../src/env';
import { disputeReadyToChallenge, SCHEME_DOWN, submitChallenge } from './account-profiles';

test.skip(syntheticAccounts().length < 2 || !hasOwner(), 'needs DM_OWNER and two DM_SYNTHETIC_ACCOUNTS');

// (9a) Per-account flow profiles. The owner assigns this worker's account
// (the first in DM_SYNTHETIC_ACCOUNTS) a profile that fails every
// challenge, in worker setup, and the worker restores the old assignment
// when it ends. A synthetic account cannot choose a flow itself (its
// dm_flow cookie is ignored); it gets what the owner assigned. The other
// half (23-account-profile-platform) runs at the same time as another
// account and sees the normal path. Tagged @account-profiles: concurrent
// CI jobs share the account pool, so only one of them runs the pair
// (`--grep-invert @account-profiles` in the others).
test.use({ accountSlot: 0, accountFlow: SCHEME_DOWN });

test('@account-profiles an account assigned a failing profile sees the scheme outage', async ({ page, dm }) => {
  const { ref } = await dm.flowProfile(SCHEME_DOWN.name, SCHEME_DOWN.variants);
  const disputeId = await disputeReadyToChallenge(dm);
  const mark = await dm.mark();

  await submitChallenge(page, disputeId);
  await expect(page.getByText('The card scheme is unavailable; the action was not applied. Retry later.')).toBeVisible();

  const [row] = await dm.waitFor({ after_seq: mark, correlation_id: dm.correlationId, action: 'dispute.challenge' });
  expect(row).toMatchObject({
    outcome: 'failed',
    error_code: 'simulated_scheme_unavailable',
    actor: { type: 'synthetic_account' },
    request: { source: 'portal' },
    target: { id: disputeId },
    flow: { profile_id: ref, profile_source: 'account' },
  });
  expect((await dm.api.get<{ status: string }>(`/v1/disputes/${disputeId}`)).status).toBe('evidence_required');
});

When a restricted-role test needs a variant that differs per test on the same account, split it: create the resource with an owner-minted key under X-DM-Flow so the variant is stamped on it, then drive it from the restricted browser; use expire-sessions for expiry; and test request-scope failure UX in an owner context.

In the portal

• Team › Synthetic accounts: the create and edit dialogs have a Flow profileselect (Workspace default, Platform behaviour, each profile as “always the latest version” and each version pinned). The list shows a badge on every account with its own profile, and a warning badge while an archived profile is in its 24-hour grace.
• Settings › Flows: each profile shows “Used by N synthetic accounts” (linking to Team, filtered) and a “Workspace default” tag; archiving one in use warns with the count and the 24-hour fallback.
• Dev panel › Flow defaults: switch the workspace default, or edit its values (saved as a new version, then the default moves to it). It is the same setting as Settings › Flows, through the same API, and says how many accounts use their own profile and are not affected.

All of it is shown to the owner only.

Every wait a variant introduces (a report's delay, a response window, a review, an acknowledgement, evidence processing, a held webhook) and every chunk of a generate job is a scheduled job (sj_…): a database row that survives deploys and restarts and runs at run_at. A test either waits for it for real or moves it forward.

States and reading them

scheduled → running → succeeded, failed or canceled. A job whose server was interrupted goes back to scheduled by itself and is retried, up to the job's max_attempts; one that runs out of attempts fails with job_lease_expired and puts its target in its failed state. Finished jobs are kept 7 days (failed: 30).

Fast-forward

POST /v1/scheduled-jobs/fast-forward takes exactly one selector, optional narrowing and one amount:

• The timeline moves with the job, in the same transaction: a report's ready_at, a dispute's respond_by and review end, a pending action's completes_at, an evidence file's processing_until. The job and the resource record fast_forwarded_by.
• Moved jobs go ahead of the workspace's other due work.
• The response is {count, matched: [{id, kind, target, state, run_at}], terminal}: jobs already running or finished are listed with their state instead of silently missing. At most 500 may match (else 400 too_many_matches); another workspace's ids match nothing. Recorded as scheduled_job.fast_forward.
• POST /v1/dev/advance-lifecycle with as_ofalso fast-forwards the targeted disputes' scheduled lifecycle jobs due by then.
• Simulated webhook retries (fail_first_n) follow the normal retry schedule; fast-forward does not move them.

Cancel

POST /v1/scheduled-jobs/:id/cancel cancels one job; POST /v1/scheduled-jobs/cancel takes the same selectors and narrowing as fast-forward (bulk, at most 500). Only scheduled and running jobs are canceled; finished ones are listed with their state and count says how many were canceled. Recorded as scheduled_job.cancel. What happens to the target depends on the kind:

The example's global teardown cancels every job its run left behind, by the run's correlation prefix:

export async function cancelRun(api: DmApi): Promise<number> {
  const pending = await listScheduledJobs(api, { correlation_id: RUN_CORRELATION_FILTER, state: 'scheduled' });
  let canceled = 0;
  for (let i = 0; i < pending.length; i += 200) {
    const job_ids = pending.slice(i, i + 200).map((j) => j.id);
    const res = await api.post<ScheduledJobBulkResult>('/v1/scheduled-jobs/cancel', { job_ids });
    canceled += res.count;
  }
  return canceled;
}

How late a job may start

A due light job (deadlines, rulings, acknowledgements, evidence, webhook releases), including one you just fast-forwarded, starts within 5 seconds at the 99th percentile while the workspace has fewer than 200 due jobs, and within 15 seconds during a deploy. Heavy jobs (report builds, generate chunks) run one at a time per server. Each job reports its own lateness_ms.

Only non-default variants schedule work, so these never apply to a workspace that does not use flows. GET /v1/flows/limitslists every cap with the workspace's current usage and, when you send X-Correlation-Id, the usage of your correlation prefix.

• Per-run sub-caps. One correlation prefix (the part of X-Correlation-Id before the first :) may hold at most half of each job cap (prefix_max in the response), so one CI run can never exhaust the workspace for the others.
• Freeing slots. A job holds its slot only while it is scheduled or running. A canceled job, and a fast-forwarded one once it has run, frees its slot at once; so does a deleted report. Cancel what your run no longer needs instead of letting it wait out its delay.

{"correlation_prefix":"run-4812",
 "data":[{"key":"pending_report_builds","description":"Report builds waiting or running.",
          "per":"tenant","min":null,"max":20,"unit":"jobs","prefix_max":10,
          "usage":{"tenant":3,"correlation_prefix":1},"error_code":"too_many_scheduled_jobs"}, …]}

Every audit row an applied variant changed carries a flow block, and it is a list and /wait filter (flow_profile_id=fp_…, with or without @N):

"flow": {
  "profile_id": "fp_01J9Z3…@2",
  "profile_source": "header",            // header | account | tenant_default
  "variants": {                          // only the non-default keys that changed this action
    "dispute.action_failure": {"value": {"actions":["challenge"],"mode":"always","error":"scheme_unavailable"},
                               "source": "request"}  // request | stamped
  }
}

• flow.profile_source says where the profile came from: header (X-DM-Flow), account(the synthetic account's assigned profile) or tenant_default. Each variant's source says whether it was read from this request (request) or from the resource's snapshot (stamped).
• Under fp_platform (and for everything a variant did not change) the row has no flow block; platformappears only as a snapshot's source on resources.
• Simulated failures are rows with outcome: "failed" and a simulated_* error_code (simulated_scheme_unavailable, simulated_service_unavailable, simulated_rate_limited…), so a test asserts on the failure it asked for and never confuses it with a real one. A simulated timeout after commit writes the normal success row with metadata.simulated_timeout: true.
• Work a scheduled job does later (report.build, evidence.process, dispute.complete_action, dispute.rule, dispute.escalate…) is a system row under the correlation id of the request that scheduled it, so correlation_id=<yours> finds it.
• Flow management itself is audited: flow_profile.create, flow_profile.update (a per-key diff), flow_profile.delete, tenant_flow_default.update, scheduled_job.fast_forward, scheduled_job.cancel, scheduled_job.expire, synthetic_account.expire_sessions and synthetic_account.flow_profile_released. See the action catalogue.

import { randomUUID } from 'node:crypto';
import { expect, test } from '../../src/dm';
import { uploadEvidence } from '../../src/evidence';
import type { Dispute, DisputePage, ErrorBody } from '../../src/types';

// (3) "Did it land?": the challenge is committed, then the API answers
// 504 as if the connection dropped. A client that retries with the same
// Idempotency-Key gets the original success back, and the dispute changed
// status exactly once.
test('a timed-out challenge retried with the same Idempotency-Key changes the status once', async ({ dm }) => {
  const { job } = await dm.generate({ count: 12, dispute_rate: 1, seed: 42 });
  const page = await dm.api.get<DisputePage>('/v1/disputes', {
    generate_job_id: job.id,
    status: 'evidence_required',
    page_size: 1,
  });
  expect(page.items).toHaveLength(1);
  const dispute = page.items[0];
  await uploadEvidence(dm, dispute.id);

  const timeout = await dm.flowProfile('challenge-timeout', {
    'dispute.action_failure': { actions: ['challenge'], mode: 'timeout_after_commit' },
  });
  await dm.useFlow(timeout.ref);
  const mark = await dm.mark();
  const key = randomUUID();
  const call = { data: { note: 'see the receipt' }, headers: { 'Idempotency-Key': key } };

  const first = await dm.api.fetch<ErrorBody>('POST', `/v1/disputes/${dispute.id}/challenge`, call);
  expect(first.status).toBe(504);
  expect(first.body.code).toBe('simulated_timeout');

  const retry = await dm.api.fetch<Dispute>('POST', `/v1/disputes/${dispute.id}/challenge`, call);
  expect(retry.status).toBe(201);
  expect(retry.headers['idempotent-replayed']).toBe('true');
  expect(retry.body).toMatchObject({ id: dispute.id, status: 'under_review' });

  const challenges = await dm.expectSequence(
    { after_seq: mark, correlation_id: dm.correlationId, action: 'dispute.challenge' },
    [
      {
        action: 'dispute.challenge',
        outcome: 'success',
        target: { id: dispute.id },
        changes: { status: { from: 'evidence_required', to: 'under_review' } },
        metadata: { simulated_timeout: true },
        flow: { profile_id: timeout.ref, profile_source: 'header' },
      },
    ],
    { mode: 'exact' },
  );
  expect(challenges).toHaveLength(1);
});

import { expect, test } from '../../src/dm';
import { uploadEvidence } from '../../src/evidence';
import type { DisputePage, ErrorBody, Evidence } from '../../src/types';

// (4) A virus scan that takes half an hour and then rejects the file. While
// the file is processing, a challenge is refused (it would submit nothing);
// once it is rejected the dispute has no usable evidence at all.
test('evidence processing then rejected blocks the challenge', async ({ dm }) => {
  const { job } = await dm.generate({ count: 12, dispute_rate: 1, seed: 42 });
  const page = await dm.api.get<DisputePage>('/v1/disputes', {
    generate_job_id: job.id,
    status: 'evidence_required',
    page_size: 1,
  });
  expect(page.items).toHaveLength(1);
  const dispute = page.items[0];

  const scan = await dm.flowProfile('evidence-virus-30m', {
    'evidence.processing': { delay_s: 1800, outcome: 'rejected', reason: 'virus_detected' },
  });
  await dm.useFlow(scan.ref);
  const mark = await dm.mark();
  const evidence = await uploadEvidence(dm, dispute.id);
  expect(evidence).toMatchObject({ status: 'processing', flow: { source: 'header' } });

  const early = await dm.api.fetch<ErrorBody>('POST', `/v1/disputes/${dispute.id}/challenge`, { data: {} });
  expect(early.status).toBe(409);
  expect(early.body).toMatchObject({ code: 'evidence_processing', details: { processing_count: 1 } });

  const moved = await dm.fastForward({ target_id: evidence.id });
  expect(moved.jobs.map((j) => [j.kind, j.state])).toEqual([['evidence.process', 'succeeded']]);
  const [processed] = await dm.waitFor({ after_seq: mark, action: 'evidence.process', target_id: evidence.id });
  expect(processed).toMatchObject({
    outcome: 'success',
    changes: { status: { from: 'processing', to: 'rejected' } },
    metadata: { rejection_reason: 'virus_detected', file_deleted: true },
  });

  const after = await dm.api.get<{ evidence: Evidence[] }>(`/v1/disputes/${dispute.id}`);
  expect(after.evidence).toEqual([expect.objectContaining({ id: evidence.id, status: 'rejected', rejection_reason: 'virus_detected' })]);
  const late = await dm.api.fetch<ErrorBody>('POST', `/v1/disputes/${dispute.id}/challenge`, { data: {} });
  expect(late.status).toBe(400);
  expect(late.body.code).toBe('evidence_missing');
  const download = await dm.api.fetch<ErrorBody>('GET', `/v1/disputes/${dispute.id}/evidence/${evidence.id}/download-url`);
  expect(download.status).toBe(409);
  expect(download.body).toMatchObject({ code: 'evidence_not_available', details: { status: 'rejected' } });
});

A key holds the scopes it was given that its creator still holds, so only keys the owner minted can carry a flow scope. Every other role gets 403 missing_permission on every flow route, and no synthetic account role can be given flow.read or flow.write.

expire-sessions vs revoke-sessions

For sessions that always expire after N seconds, assign the account a profile with session.max_age instead.