Use this workflow when recruiting operations needs to know which role or stage merits a closer look. The download includes an Ashby collector, a PostgreSQL schema and an inactive n8n workflow. Its synthetic example produces four aggregate alerts. You can inspect the evidence behind each one before connecting any accounts.
Run the example
Extract the download and run python3 -B check_fixture.py with Python 3.10 or later. Then print the snapshot that feeds both detectors:
The date is the exclusive UTC cutoff. The daily window covers November 1 and the weekly window starts October 26. The fixture contains two pages of applications for the Analyst role and two history pages for one application. It also includes a watched role with no applications and a recently added role. The adapter reads ten synthetic response pages and carries all three roles into the snapshot.
Expected alert
Evidence in the fixture
Stage conversion drop
One of four completed Screen departures advanced to Onsite. The 25% rate falls below the configured 75% reference mean minus two deviations of 10 percentage points.
Completed stage dwell
The median completed Screen interval was 48 hours. The alert threshold is 1.5 times the configured 24-hour limit.
Quiet role
The watched Quiet role had zero entries or exits in seven days. The newer role had fewer than seven monitored days and produced no alert.
Hire trend
One application reached the configured hire stage 59 days after creation, above the example 30-day threshold. The fixture’s minimum hire count is one.
fixtures/expected-alerts.json contains the expected payloads. For one alert, run node check_postgres.mjs --one-alert after the PGlite setup in _README.md. Its quiet-role scenario queues one alert on the first run and zero new alerts on the second. The SQL check runs the hire detector first, then the stage detector. It gets one new hire alert and three new stage/role alerts. Running both again creates zero additional alerts. The example’s thresholds and small sample sizes illustrate the rules. Choose reference values and minimum samples that fit your own funnel metrics.
Prepare the source and database
Copy config.example.json and replace every invented job ID, stage ID, label, monitoring date, reference value and limit. Define which successor stages count as advancement. Calculate reference mean and deviation from eligible daily rates using the same completed-departure denominator. The rule requires at least 20 reference days and a positive deviation. Refresh dates older than 35 days are rejected. Reference backfill and refresh scheduling belong to the operator.
The separate collector calls Ashby application.list for each watched job, then application.listHistory for each application. Both are paginated POST requests using candidatesRead. The collector orders histories by stage number and uses stage IDs and timestamps. A missing page, duplicate record or inconsistent sequence aborts collection. Zero activity is meaningful only after that complete collection succeeds.
Apply schema.sql with psql, set ASHBY_API_KEY for the collector and schedule the live command in _README.md before 02:00 UTC. --persist stores one complete snapshot using the configured PostgreSQL connection. The two tables are hfa.snapshots and hfa.alerts. An identical snapshot can be loaded again. A different payload for the same date requires review. Candidate names, emails and application IDs stay out of the persisted aggregates. Review role labels before sending them to external services.
Connect the two detectors and delivery
Import hiring-funnel-anomaly-n8n.json into n8n. Bind PostgreSQL, Anthropic Header Auth and Slack Header Auth credentials. Set the approved stage and leadership channel IDs in Check channel configuration. Keep the workflow inactive until the two detectors return the expected fixture rows in a disposable database. The collector uses its own Ashby credential outside n8n.
The 02:00 UTC detector evaluates conversion, completed dwell and seven-day role activity. Conversion counts completed exits with a consecutive observed successor inside the cutoff. Terminal exits without a successor and gaps in stage order are excluded. Dwell measures completed intervals, so an alert describes a stage rather than an individual currently waiting in it. The 03:00 detector independently checks elapsed time from application creation to first entry into the configured hire stage. It counts a hire only when history is complete from ordinal zero through that hire-stage entry. A missing prefix leaves the first hire date unverified and excludes the application from this statistic. Confirm the meaning of the stage in your ATS before treating the metric as time to hire. Either detector fails visibly if that day’s snapshot is missing.
Both schedules write stable alert keys to the same outbox. A delivery tick claims one pending row after detection and every 15 minutes. Claude receives aggregate metrics for a short explanation. Slack receives the figures, explanation and reference key. Check the stored delivery_state result and the row’s state. A confirmed Slack response includes ok:true, a channel and a timestamp. A known rejection such as ok:false with channel_not_found records a failed delivery. Slack says internal_error and fatal_error can follow partial success. Those errors, unknown codes, incomplete acknowledgements and send timeouts leave acceptance uncertain. Inspect the channel for the reference before deciding whether to retry. The bundle holds uncertain sends for review. See Slack’s response contract.
Verification and operating cost
Twelve offline Python checks cover pagination, invalid inputs, date boundaries, empty roles and export structure. The shipped SQL ran on PostgreSQL 18.6 through PGlite and independently on native PostgreSQL 18.4 with synthetic records. The tests exercised both schedules, exact alert payloads and replay. The PGlite checks also covered delivery leases, confirmed acknowledgements, definite rejection and uncertainty. The unmodified export imported into n8n 2.45.7. Seven CLI runs used copies with manual triggers, local test credentials and loopback model/Slack endpoints. The actual SQL, Code and HTTP nodes processed four mock explanations and four confirmed mock posts, two per channel. Repeating the stage, hire and delivery runs added zero calls or alerts. The README gives the fixture commands. Wall-clock schedules and live provider accounts still require an operator run.
Budget for the number of applications and history pages scanned per watched role, PostgreSQL storage, n8n executions and model calls on claimed alerts. In the four-alert example, four successful first attempts would request four explanations and four Slack posts. Pre-send retries can add model usage. Measure actual token counts, delivery delay and reviewer time in your environment before committing to a cost or response target.
Use a spreadsheet or your existing ATS report when nobody owns the reference values, routing or follow-up. Revisit the detector when stable stage definitions and a named reviewer are in place. The output supports investigation of aggregate recruiting operations. The graph contains no ATS mutation step.
# Hiring funnel detector
Start with the synthetic fixture. It needs Python 3.10+ and runs offline. Run these commands in the extracted bundle directory.
```sh
python3 -B check_fixture.py
python3 -B ashby_adapter.py --config config.example.json --fixture fixtures/ashby-pages.json --as-of 2026-11-02
```
The first command checks pagination, missing data, duplicates, timestamp boundaries, role coverage and the export's two detection paths. The second prints the aggregates consumed by PostgreSQL. Five invented applications belong to the Analyst role. Four left Screen with an observed successor during November 1 UTC, one advanced to Onsite and the median completed dwell was 48 hours. A separate application first reached the configured hire stage 59 days after creation. A watched role has zero activity. Another role was added too recently to qualify for the seven-day rule.
`fixtures/expected-alerts.json` gives the four expected alert payloads. `check_postgres.mjs` starts a disposable PostgreSQL engine via PGlite, applies the actual `schema.sql`, runs the hire branch before the stage branch, and checks those payloads, snapshot immutability, replay, samples and delivery states. Install the pinned test dependency in the extracted directory and run the SQL checks.
```sh
npm install --prefix . --no-save --package-lock=false --ignore-scripts --no-audit --no-fund @electric-sql/pglite@0.5.14
node check_postgres.mjs
node check_postgres.mjs --one-alert
```
Alternatively, set `HFA_PGLITE_MODULE` to the installed package's absolute `dist/index.js` path. For the smaller example, run `node check_postgres.mjs --one-alert`. It uses `fixtures/one-alert-config.json` in a fresh, isolated database. The first run queues one quiet-role alert. A second run queues zero new alerts, and one pending row remains. The complete fixture ran on PostgreSQL 18.6 via PGlite 0.5.14. Independent native PostgreSQL 18.4 execution also produced three stage alerts and one hire alert, followed by zero new alerts on replay. The Python and SQL checks use synthetic responses.
The unmodified export was also imported into n8n 2.45.7. Seven CLI runs used disposable copies with manual CLI triggers, local test credentials, test channel IDs and loopback mock URLs for Anthropic and Slack. With native PostgreSQL 18.4, the stage run queued three alerts and sent one, the hire run queued one and sent one, and two outbox runs sent the remaining two. Repeating stage, hire and drain left four delivered rows, four mock model requests and four mock Slack acknowledgements. Two alerts routed to the stage channel and two to leadership. These runs exercise the n8n SQL, Code and HTTP nodes using synthetic records. Wall-clock trigger firing and live Ashby, Anthropic and Slack services require a separate operator check.
## Data contract
`ashby_adapter.py` is a separate collector. The n8n export reads its completed snapshot. Schedule the collector before 02:00 UTC. Use a read-only Ashby API key with `candidatesRead`, exported as `ASHBY_API_KEY`. The collector sends Basic authentication with the key as username and an empty password, and pins `Ashby-Version: 2026-01-01`. Keep the key out of configuration, command arguments and fixtures.
For each explicitly watched job, the collector posts `jobId`, `createdBefore`, `cursor` and `limit` to `application.list`. It retrieves all pages and posts each `applicationId` to `application.listHistory`, following that endpoint's own pagination. `stageNumber` orders the stage records. `stageId`, `enteredStageAt` and `leftStageAt` identify transitions and durations. The collector uses `createdAt` from the application to measure elapsed time to the first configured hire-stage entry. See [Ashby application list](https://developers.ashbyhq.com/reference/applicationlist), [stage history](https://developers.ashbyhq.com/reference/applicationlisthistory) and [API versioning](https://developers.ashbyhq.com/reference/versioning).
This is a bounded full scan per watched job, with a maximum of 5,000 applications per job and 100 pages per list. Estimate request volume from the number of applications and history pages. No incremental checkpoint or Greenhouse/Lever adapter is included. A failed request, inconsistent job, broken pagination, duplicate record or invalid stage sequence aborts the entire collection before a snapshot is written. Review the error and rerun the same date. An empty but successful job response retains a zero row for its configured role.
Copy `config.example.json`. Replace every invented ID, date, label, advancing-successor list and reference value. Use the stage IDs in your Ashby account. `monitor_since` is the earliest date on which you can vouch for coverage of that role. Choose neutral aggregate labels that may be sent to Claude and Slack. The collector reads applications in memory. It persists role/stage aggregates, counts and policy values; the output contains no application ID, candidate name, email or source transcript.
The `reference_mean`, `reference_sd` and `reference_days` inputs are operator-supplied. For a comparable historical period, calculate each eligible day's advancing exits divided by completed exits with an observed successor, then compute the mean and standard deviation of those daily rates. Record the number of eligible days and reference update date. Use the same stage IDs, opportunity definition and daily minimum as the live rule. A taxonomy change needs a new reference. The adapter rejects a reference updated in the future or more than 35 days before the snapshot date. The example thresholds are policy choices, not population benchmarks. Reference backfill and refresh scheduling are your responsibility.
## PostgreSQL and schedules
Use a database approved for these aggregates and the standard `psql` client. Configure the appropriate `PGHOST`, `PGPORT`, `PGDATABASE`, `PGUSER` and credential mechanism. Apply the schema in that database, then load a fixture into a disposable database or collect real data into the configured destination:
```sh
psql -X -v ON_ERROR_STOP=1 -f schema.sql
# Disposable fixture only
python3 -B ashby_adapter.py --config config.example.json --fixture fixtures/ashby-pages.json --as-of 2026-11-02 --persist
# Operator-configured live collector; replace the date with today's completed UTC cutoff
python3 -B ashby_adapter.py --config watched-roles.json --as-of YYYY-MM-DD --persist
```
`as_of` is the UTC midnight **ending** the window. For `2026-11-02`, the daily interval is `[Nov 1 00:00, Nov 2 00:00)` and the weekly interval starts October 26. A single immutable snapshot contains all watched roles for that date. Identical replay is accepted. A changed replay raises an error so a correction cannot silently rewrite evidence for delivered alerts. `hfa.snapshots` and `hfa.alerts` are the complete table requirements. The hire trend comes from the snapshot, without a separate unprovided `hires` table.
Import `hiring-funnel-anomaly-n8n.json` into n8n. It ships inactive, with `executionOrder: v1` and timezone `UTC`. Bind the PostgreSQL nodes to the database with the `hfa` schema. Bind the Anthropic HTTP node to Header Auth `x-api-key`. Bind the Slack HTTP node to Header Auth `Authorization` with your bot bearer token. Invite the bot to both channels and grant `chat:write`. In **Check channel configuration**, replace `REPLACE_STAGE_CHANNEL_ID` and `REPLACE_LEADERSHIP_CHANNEL_ID` with approved channel IDs before enabling the schedules. The Claude request uses aggregate alert data, model `claude-haiku-5-5` and a 120-token output cap. Verify current model access and prices in your account. See [Anthropic models](https://platform.claude.com/docs/en/models/overview) and [Slack chat.postMessage](https://api.slack.com/methods/chat.postMessage).
The 02:00 UTC schedule calls `hfa.detect_stages(date)`. The 03:00 schedule calls `hfa.detect_tth(date)` independently. Both require that exact day's completed snapshot. An absent snapshot raises an error. After correcting a failed collection, rerun the relevant detector for the original date using SQL or the node's date parameter. Confirm the calls in a disposable database:
```sql
SELECT hfa.detect_tth('2026-11-02'); -- 1 new alert on the fixture
SELECT hfa.detect_stages('2026-11-02'); -- 3 new alerts on the fixture
SELECT hfa.detect_tth('2026-11-02'); -- 0 on replay
SELECT hfa.detect_stages('2026-11-02'); -- 0 on replay
SELECT payload, state FROM hfa.alerts ORDER BY payload->>'kind';
```
## Rules and delivery
- **Stage conversion.** Compare the observed share of completed departures that reached a configured advancing successor with the historical daily mean. The rule requires the configured minimum daily departures, at least 20 reference days and a positive reference deviation. It fires at or below `mean - 2 × standard deviation`. Pending candidates, terminal exits without a successor, gaps in stage ordering and successors starting after the cutoff are outside this denominator. Nonzero starting stage numbers are supported. Duplicate or invalid order values abort collection.
- **Completed dwell.** With the same daily sample gate, flag a median completed dwell above `1.5 × dwell_limit_hours`. This describes completed transitions. It does not identify individual candidates who are currently stuck.
- **Quiet role.** Flag a configured role after at least seven monitored days when no stage-history row has an entry or exit in the seven-day window. Empty application lists qualify only after successful collection. A new role with fewer than seven monitored days is suppressed.
- **Hire trend.** Average days from application creation to first entry into the configured hire stage for hires within the seven-day window. Flag when the average exceeds `tth_limit_days` and the configured minimum hire count is met. Count a hire only when its observed history has a contiguous ordinal prefix from zero through the first configured hire-stage entry. A missing prefix leaves the first hire date unverified and excludes that application from the hire statistic. Confirm the chosen stage means an actual hire in your taxonomy.
Each alert key includes UTC date, job, stage and type. Both detectors enqueue rows through the same SQL function. Reruns generate zero additional rows for an existing key. A delivery tick runs every 15 minutes and after either detector. Each tick claims at most one pending alert with a 15-minute lease. Watch the oldest pending row and queue depth if you expect more alerts than this schedule can drain.
The shared delivery path checks the channel, requests a short aggregate explanation, records that a Slack send is about to start, sends the message, and persists the response. A pre-send failure leaves a lease that can be retried. Three expired pre-send attempts mark the row failed for operator review. Repeated explanations can incur additional model usage. Slack `ok:true` with a channel and timestamp marks delivery confirmed. `ok:false` with `channel_not_found`, `not_in_channel`, `invalid_auth`, `missing_scope`, `no_text` or `msg_too_long` records a definite rejection. Other error codes and incomplete acknowledgements stay uncertain. Slack documents that `internal_error` and `fatal_error` can follow a partially successful operation. A timeout after the send starts also leaves an uncertain outcome when its lease expires. That outcome is never automatically reposted. Check the channel for the message's `Ref` before deciding how to reconcile the row. A DB transaction cannot make a remote Slack call exactly-once.
Inspect `state`, `last_error`, `slack_channel` and `slack_ts` in `hfa.alerts`. Correct a definite rejection before explicitly returning that row to `pending` with its lease fields cleared and attempts reset. For an uncertain send, first establish whether Slack accepted the message. Keep the row held if you cannot establish the outcome. Test your approved account bindings with aggregate-only sandbox data before activating production schedules. Record actual API requests, tokens, execution counts, alert acknowledgement and human review latency in your own environment.
#!/usr/bin/env python3
"""Make one complete UTC recruiting snapshot from Ashby or the bundled synthetic pages."""
import argparse
import base64
from collections import defaultdict
from datetime import date, datetime, timedelta, timezone
import json
import math
import os
from pathlib import Path
from statistics import median
import subprocess
import sys
from urllib import error, request
from uuid import UUID
HERE = Path(__file__).resolve().parent
UTC = timezone.utc
def require(condition, message):
if not condition:
raise ValueError(message)
def stamp(value):
require(isinstance(value, str), 'Timestamp must be a string')
result = datetime.fromisoformat(value.replace('Z', '+00:00'))
require(result.utcoffset() is not None, 'Timestamp requires a UTC offset')
return result.astimezone(UTC)
def number(row, key, low, high=None):
value = row[key]
require(type(value) in (int, float) and math.isfinite(value) and value >= low and
(high is None or value <= high), f'Invalid {key}')
def count(row, key, minimum):
# These values are read as PostgreSQL integer by both scheduled detectors.
require(type(row[key]) is int and minimum <= row[key] <= 2147483647, f'Invalid integer {key}')
def validate_config(config, day):
roles = config['roles']
require(isinstance(roles, list) and roles, 'Configure at least one watched role')
require(len({r['job_id'] for r in roles}) == len(roles), 'Duplicate watched role')
for role in roles:
UUID(role['job_id']); UUID(role['hire_stage_id'])
require(isinstance(role['label'], str) and role['label'].strip(), 'Missing role label')
require(date.fromisoformat(role['monitor_since']) <= day, 'Role monitoring starts in the future')
age = (day - date.fromisoformat(role['reference_updated_on'])).days
require(0 <= age <= 35, 'Refresh reference metrics before collecting this date (35-day limit)')
number(role, 'tth_limit_days', 0.01)
count(role, 'min_weekly_hires', 1)
rules = role['stages']
require(isinstance(rules, list) and rules, 'Configure at least one stage')
require(len({s['stage_id'] for s in rules}) == len(rules), 'Duplicate watched stage')
for rule in rules:
UUID(rule['stage_id'])
require(isinstance(rule['advance_to'], list) and rule['advance_to'], 'Specify advancing successor stages')
for stage in rule['advance_to']:
UUID(stage)
require(isinstance(rule['label'], str) and rule['label'].strip(), 'Missing stage label')
number(rule, 'reference_mean', 0, 1); number(rule, 'reference_sd', 0, 1)
count(rule, 'reference_days', 0); count(rule, 'min_daily_exits', 1)
number(rule, 'dwell_limit_hours', 0.01)
class Ashby:
def __init__(self, key):
require(bool(key), 'Set ASHBY_API_KEY or select --fixture')
self.auth = 'Basic ' + base64.b64encode((key + ':').encode()).decode()
def __call__(self, endpoint, body):
req = request.Request('https://api.ashbyhq.com/' + endpoint, method='POST',
data=json.dumps(body).encode(), headers={
'Authorization': self.auth, 'Content-Type': 'application/json',
'Ashby-Version': '2026-01-01'})
try:
with request.urlopen(req, timeout=30) as response:
return json.load(response)
except error.HTTPError as exc:
raise ValueError(f'Ashby {endpoint} returned HTTP {exc.code}; snapshot aborted') from None
class Fixture:
def __init__(self, pages):
self.pages = pages
def __call__(self, endpoint, body):
owner = body.get('jobId') or body.get('applicationId')
return self.pages[endpoint][owner][body['cursor']]
def paginate(fetch, endpoint, filters):
cursor = 'start'
cursors = set()
while True:
require(cursor not in cursors and len(cursors) < 100, 'Missing/repeated cursor or page limit reached')
cursors.add(cursor)
page = fetch(endpoint, dict(filters, cursor=cursor, limit=100))
require(page.get('success') is True and isinstance(page.get('results'), list) and
type(page.get('moreDataAvailable')) is bool, f'Incomplete {endpoint} response')
yield from page['results']
if not page['moreDataAvailable']:
return
cursor = page.get('nextCursor')
require(isinstance(cursor, str) and cursor, f'Missing nextCursor in {endpoint}')
def role_snapshot(fetch, rule, end):
day_start = end - timedelta(days=1)
week_start = end - timedelta(days=7)
counts = defaultdict(lambda: {'exits': 0, 'advanced': 0, 'hours': []})
weekly_moves = 0
hire_days = []
seen_apps = set()
filters = {'jobId': rule['job_id'], 'createdBefore': int(end.timestamp()*1000)-1}
for app in paginate(fetch, 'application.list', filters):
app_id = app['id']
require(app['job']['id'] == rule['job_id'], 'Application returned for a different watched job')
require(app_id not in seen_apps and len(seen_apps) < 5000, 'Duplicate application or per-role limit reached')
seen_apps.add(app_id)
created = stamp(app['createdAt'])
require(created < end, 'Application falls outside snapshot cutoff')
history = sorted(paginate(fetch, 'application.listHistory', {'applicationId': app_id}),
key=lambda e: e['stageNumber'])
require(len({e['id'] for e in history}) == len(history), 'Duplicate history event')
positions = [e['stageNumber'] for e in history]
require(all(type(n) is int and n >= 0 for n in positions) and len(set(positions)) == len(positions),
'Invalid or duplicate history order')
first_hire = None
complete_prefix = True
previous_entered = None
for position, event in enumerate(history):
complete_prefix = complete_prefix and event['stageNumber'] == position
stage = event['stageId']
entered = stamp(event['enteredStageAt'])
left = stamp(event['leftStageAt']) if event.get('leftStageAt') else None
require(previous_entered is None or entered >= previous_entered, 'Out-of-order stage history')
require(left is None or left >= entered, 'Negative stage duration')
previous_entered = entered
if entered >= end:
continue
if week_start <= entered or (left and week_start <= left < end):
weekly_moves += 1
if complete_prefix and stage == rule['hire_stage_id'] and first_hire is None:
first_hire = entered
if left and day_start <= left < end and position + 1 < len(history):
successor = history[position+1]
successor_entered = stamp(successor['enteredStageAt'])
require(left <= successor_entered, 'Overlapping stage history')
# A numbering gap does not establish which stage followed this exit.
if successor['stageNumber'] != event['stageNumber'] + 1 or successor_entered >= end:
continue
for s in rule['stages']:
if s['stage_id'] == stage:
count = counts[stage]
count['exits'] += 1
count['advanced'] += int(successor['stageId'] in s['advance_to'])
count['hours'].append((left-entered).total_seconds()/3600)
if first_hire is not None and week_start <= first_hire < end:
require(first_hire >= created, 'Hire stage precedes application creation')
hire_days.append((first_hire-created).total_seconds()/86400)
stages = []
for s in rule['stages']:
c = counts[s['stage_id']]
stages.append(dict(s, exits=c['exits'], advanced=c['advanced'],
median_dwell_hours=median(c['hours']) if c['hours'] else None))
return dict(job_id=rule['job_id'], label=rule['label'], monitor_since=rule['monitor_since'],
weekly_movement_events=weekly_moves, applications_scanned=len(seen_apps), stages=stages,
hires=len(hire_days), mean_hire_days=sum(hire_days)/len(hire_days) if hire_days else None,
tth_limit_days=rule['tth_limit_days'], min_weekly_hires=rule['min_weekly_hires'])
def collect(config, day, fetch):
validate_config(config, day)
end = datetime.combine(day, datetime.min.time(), UTC)
return {'version': 1, 'as_of': day.isoformat(),
'roles': [role_snapshot(fetch, role, end) for role in config['roles']]}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--config', type=Path, required=True)
parser.add_argument('--as-of', type=date.fromisoformat, required=True, help='UTC midnight ending the completed window')
parser.add_argument('--fixture', type=Path, help='Read synthetic API response pages instead of making requests')
parser.add_argument('--persist', action='store_true', help='Store aggregate snapshot using psql and its PG* environment')
args = parser.parse_args()
require(args.as_of <= datetime.now(UTC).date(), 'Select a completed UTC window')
fetch = Fixture(json.loads(args.fixture.read_text())) if args.fixture else Ashby(os.environ.get('ASHBY_API_KEY'))
snapshot = collect(json.loads(args.config.read_text()), args.as_of, fetch)
result = json.dumps(snapshot, sort_keys=True, allow_nan=False, ensure_ascii=False)
if args.persist:
# A psql variable quoted by :'snapshot' is a SQL literal, never executable input.
subprocess.run(['psql', '-X', '-v', 'ON_ERROR_STOP=1', '-v', 'snapshot='+result],
input="SELECT hfa.store_snapshot(:'snapshot'::jsonb);\n", text=True, check=True)
else:
print(json.dumps(snapshot, indent=2, allow_nan=False, ensure_ascii=False))
if __name__ == '__main__':
try:
main()
except (ValueError, KeyError, OSError, subprocess.SubprocessError) as exc:
sys.exit(f'Collection failed: {exc}')
#!/usr/bin/env python3
"""Offline regression checks for the actual adapter and importable graph."""
from copy import deepcopy
from datetime import date
import json
from unittest import TestCase, main
from ashby_adapter import collect, Fixture, HERE, validate_config
DAY = date(2026, 11, 2)
CONFIG = json.loads((HERE/'config.example.json').read_text())
PAGES = json.loads((HERE/'fixtures/ashby-pages.json').read_text())
JOB = CONFIG['roles'][0]['job_id']
APP = PAGES['application.list'][JOB]['start']['results'][0]['id']
class AdapterChecks(TestCase):
def test_paginated_aggregate_and_empty_roles(self):
calls = []
def fetch(endpoint, body):
calls.append((endpoint, body))
return Fixture(PAGES)(endpoint, body)
active, quiet, recent = collect(CONFIG, DAY, fetch)['roles']
stage = active['stages'][0]
self.assertEqual((active['applications_scanned'],stage['exits'],stage['advanced'],stage['median_dwell_hours']), (5,4,1,48))
self.assertEqual((active['hires'],active['mean_hire_days']), (1,59))
self.assertEqual([r['weekly_movement_events'] for r in (quiet,recent)], [0,0])
self.assertTrue(any(b['cursor']=='apps-2' for _,b in calls))
self.assertTrue(any(b['cursor']=='history-2' for _,b in calls))
self.assertFalse(any('syncToken' in b for _,b in calls))
self.assertEqual(len(calls), 10)
def test_bad_page_aborts_collection(self):
for change in [{'success':False},{'nextCursor':None},{'nextCursor':'start'},{'moreDataAvailable':'true'}]:
pages=deepcopy(PAGES); pages['application.list'][JOB]['start'].update(change)
with self.subTest(change=change),self.assertRaises(ValueError): collect(CONFIG,DAY,Fixture(pages))
def test_duplicate_app_does_not_inflate_denominator(self):
pages=deepcopy(PAGES)
pages['application.list'][JOB]['apps-2']['results'].append(pages['application.list'][JOB]['start']['results'][0])
with self.assertRaisesRegex(ValueError,'Duplicate application'): collect(CONFIG,DAY,Fixture(pages))
def test_response_from_wrong_job_is_rejected(self):
pages=deepcopy(PAGES);pages['application.list'][JOB]['start']['results'][0]['job']['id']='wrong'
with self.assertRaisesRegex(ValueError,'different watched job'): collect(CONFIG,DAY,Fixture(pages))
def test_conflicting_history_is_not_treated_as_no_activity(self):
pages=deepcopy(PAGES);pages['application.listHistory'][APP]['history-2']['results'][0]['stageNumber']=0
with self.assertRaisesRegex(ValueError,'duplicate history order'): collect(CONFIG,DAY,Fixture(pages))
def test_nonzero_order_terminal_exit_and_numbering_gap(self):
pages=deepcopy(PAGES)
for name in ('start','history-2'):
pages['application.listHistory'][APP][name]['results'][0]['stageNumber'] += 4
stage=lambda: collect(CONFIG,DAY,Fixture(pages))['roles'][0]['stages'][0]
self.assertEqual((stage()['exits'],stage()['advanced']),(4,1))
pages['application.listHistory'][APP]['history-2']['results'][0]['stageNumber'] += 1
self.assertEqual((stage()['exits'],stage()['advanced']),(3,0))
pages['application.listHistory'][APP]['history-2']['results']=[]
self.assertEqual((stage()['exits'],stage()['advanced']),(3,0))
def test_right_boundary_is_exclusive(self):
pages=deepcopy(PAGES)
pages['application.listHistory'][APP]['start']['results'][0]['leftStageAt']='2026-11-02T00:00:00Z'
pages['application.listHistory'][APP]['history-2']['results'][0]['enteredStageAt']='2026-11-02T00:00:00Z'
row=collect(CONFIG,DAY,Fixture(pages))['roles'][0]['stages'][0]
self.assertEqual((row['exits'],row['advanced']), (3,0))
def test_first_hire_requires_complete_prefix(self):
pages=deepcopy(PAGES)
hire_app=pages['application.list'][JOB]['apps-2']['results'][-1]['id']
pages['application.listHistory'][hire_app]['start']['results'][0]['stageNumber']=8
row=collect(CONFIG,DAY,Fixture(pages))['roles'][0]
self.assertEqual((row['hires'],row['mean_hire_days']),(0,None))
def test_missing_or_stale_reference_fails_before_fetching(self):
bad=deepcopy(CONFIG);bad['roles'][0]['reference_updated_on']='2026-08-01'
with self.assertRaisesRegex(ValueError,'Refresh reference'): validate_config(bad,DAY)
bad=deepcopy(CONFIG);bad['roles'][0]['stages'][0]['reference_sd']=float('nan')
with self.assertRaisesRegex(ValueError,'reference_sd'):validate_config(bad,DAY)
def test_aggregate_serialization_drops_application_and_candidate_identifiers(self):
pages=deepcopy(PAGES)
pages['application.list'][JOB]['start']['results'][0]['candidate']={'name':'SYNTHETIC NAME','email':'fixture@example.invalid'}
result=json.dumps(collect(CONFIG,DAY,Fixture(pages)))
for text in (APP,'SYNTHETIC NAME','fixture@example.invalid','candidate','enteredStageAt'):
self.assertNotIn(text,result)
def test_integer_counts_fail_before_snapshot_storage(self):
for key in ['min_weekly_hires','reference_days','min_daily_exits']:
for bad_value in (1.5, True, '3', 2147483648):
config=deepcopy(CONFIG)
target=config['roles'][0] if key=='min_weekly_hires' else config['roles'][0]['stages'][0]
target[key]=bad_value
with self.subTest(key=key,value=bad_value),self.assertRaisesRegex(ValueError,'Invalid integer'):
validate_config(config,DAY)
def test_export_routes_both_schedules_through_shared_claim(self):
flow=json.loads((HERE/'hiring-funnel-anomaly-n8n.json').read_text());nodes={n['name']:n for n in flow['nodes']}
self.assertFalse(flow['active']);self.assertEqual(flow['settings']['timezone'],'UTC')
def trace(name):
edges=flow['connections'].get(name,{}).get('main',[[]])[0]
return [name]+(trace(edges[0]['node']) if edges else [])
for name in ['Stage detector at 02 UTC','Hire trend at 03 UTC','Drain outbox every 15 minutes']:
path=trace(name)
self.assertEqual(path[path.index('Claim one pending alert'):],['Claim one pending alert','Check channel configuration','Claude aggregate explanation','Prepare Slack message','Record send attempt','Post Slack message','Persist Slack acknowledgement','Check delivery state'])
self.assertEqual(nodes['Claude aggregate explanation']['parameters']['genericAuthType'],'httpHeaderAuth')
self.assertEqual(nodes['Post Slack message']['parameters']['url'],'https://slack.com/api/chat.postMessage')
if __name__=='__main__': main(verbosity=2)
-- Run once in your own PostgreSQL database, then bind the n8n credential to it.
BEGIN;
CREATE SCHEMA IF NOT EXISTS hfa;
CREATE TABLE IF NOT EXISTS hfa.snapshots (
as_of date PRIMARY KEY,
payload jsonb NOT NULL,
loaded_at timestamptz NOT NULL DEFAULT clock_timestamp()
);
CREATE TABLE IF NOT EXISTS hfa.alerts (
alert_key text PRIMARY KEY,
as_of date NOT NULL REFERENCES hfa.snapshots(as_of),
payload jsonb NOT NULL,
route text NOT NULL CHECK (route IN ('stage','leadership')),
state text NOT NULL DEFAULT 'pending' CHECK (state IN ('pending','leased','sending','delivered','failed','uncertain')),
attempts integer NOT NULL DEFAULT 0,
lease_token text, lease_until timestamptz, slack_channel text, slack_ts text, last_error text
);
CREATE OR REPLACE FUNCTION hfa.store_snapshot(p jsonb) RETURNS date LANGUAGE plpgsql AS $$
DECLARE d date := (p->>'as_of')::date;
BEGIN
IF d IS NULL OR p->>'version' IS DISTINCT FROM '1' OR jsonb_typeof(p->'roles') IS DISTINCT FROM 'array'
OR jsonb_array_length(p->'roles')=0 THEN RAISE EXCEPTION 'Invalid snapshot'; END IF;
INSERT INTO hfa.snapshots(as_of,payload) VALUES(d,p) ON CONFLICT DO NOTHING;
IF (SELECT payload FROM hfa.snapshots WHERE as_of=d) IS DISTINCT FROM p THEN
RAISE EXCEPTION 'A different snapshot already exists for %; review before changing recorded evidence', d;
END IF;
RETURN d;
END $$;
CREATE OR REPLACE FUNCTION hfa.require_snapshot(d date) RETURNS jsonb LANGUAGE plpgsql AS $$
DECLARE p jsonb;
BEGIN
SELECT payload INTO p FROM hfa.snapshots WHERE as_of=d;
IF NOT FOUND THEN RAISE EXCEPTION 'Missing complete UTC snapshot for %',d; END IF;
RETURN p;
END $$;
CREATE OR REPLACE FUNCTION hfa.enqueue(d date,r jsonb,s jsonb,kind text,reading numeric,reference numeric,unit text,n integer,channel_kind text)
RETURNS integer LANGUAGE plpgsql AS $$
DECLARE inserted integer;
BEGIN
INSERT INTO hfa.alerts(alert_key,as_of,payload,route) VALUES(
jsonb_build_array(d,r->>'job_id',s->>'stage_id',kind)::text,d,
jsonb_build_object('as_of',d,'role',r->>'label','stage',s->>'label','kind',kind,
'observed',reading,'reference',reference,'unit',unit,'sample_size',n),channel_kind)
ON CONFLICT DO NOTHING;
GET DIAGNOSTICS inserted = ROW_COUNT;
RETURN inserted;
END $$;
CREATE OR REPLACE FUNCTION hfa.detect_stages(d date) RETURNS integer LANGUAGE plpgsql AS $$
DECLARE r jsonb; s jsonb; n integer; rate numeric; total integer := 0;
BEGIN
FOR r IN SELECT value FROM jsonb_array_elements(hfa.require_snapshot(d)->'roles') LOOP
IF d-(r->>'monitor_since')::date >= 7 AND (r->>'weekly_movement_events')::int=0 THEN
total := total + hfa.enqueue(d,r,NULL,'role_no_movement',0,NULL,'movement events in 7 days',
(r->>'applications_scanned')::int,'leadership');
END IF;
FOR s IN SELECT value FROM jsonb_array_elements(r->'stages') LOOP
n := (s->>'exits')::int;
IF n < (s->>'min_daily_exits')::int THEN CONTINUE; END IF;
rate := (s->>'advanced')::numeric/n;
IF (s->>'reference_days')::int >= 20 AND (s->>'reference_sd')::numeric > 0 AND
rate <= (s->>'reference_mean')::numeric - 2*(s->>'reference_sd')::numeric THEN
total := total + hfa.enqueue(d,r,s,'stage_conversion_drop',rate,(s->>'reference_mean')::numeric,'fraction',n,'stage');
END IF;
IF (s->>'median_dwell_hours')::numeric > 1.5*(s->>'dwell_limit_hours')::numeric THEN
total := total + hfa.enqueue(d,r,s,'completed_stage_dwell', (s->>'median_dwell_hours')::numeric,
(s->>'dwell_limit_hours')::numeric,'hours',n,'stage');
END IF;
END LOOP;
END LOOP;
RETURN total;
END $$;
CREATE OR REPLACE FUNCTION hfa.detect_tth(d date) RETURNS integer LANGUAGE plpgsql AS $$
DECLARE r jsonb; total integer := 0;
BEGIN
FOR r IN SELECT value FROM jsonb_array_elements(hfa.require_snapshot(d)->'roles') LOOP
IF (r->>'hires')::int >= (r->>'min_weekly_hires')::int AND
(r->>'mean_hire_days')::numeric > (r->>'tth_limit_days')::numeric THEN
total := total + hfa.enqueue(d,r,NULL,'time_to_hire_trend', (r->>'mean_hire_days')::numeric,
(r->>'tth_limit_days')::numeric,'days',(r->>'hires')::int,'leadership');
END IF;
END LOOP;
RETURN total;
END $$;
CREATE OR REPLACE FUNCTION hfa.claim_pending(token text) RETURNS SETOF hfa.alerts LANGUAGE plpgsql AS $$
BEGIN
IF NULLIF(token,'') IS NULL THEN RAISE EXCEPTION 'Missing execution token'; END IF;
-- A Slack request could already have been accepted. Never automatically repost it.
UPDATE hfa.alerts SET state='uncertain',last_error='No definitive Slack response recorded'
WHERE state='sending' AND lease_until <= clock_timestamp();
UPDATE hfa.alerts SET state='failed',last_error='Three pre-send leases expired; inspect before retrying'
WHERE state='leased' AND attempts>=3 AND lease_until<=clock_timestamp();
RETURN QUERY WITH picked AS (
SELECT a.alert_key FROM hfa.alerts a WHERE a.state='pending' OR (a.state='leased' AND a.lease_until<=clock_timestamp())
ORDER BY a.as_of,a.alert_key LIMIT 1 FOR UPDATE SKIP LOCKED
) UPDATE hfa.alerts a SET state='leased',lease_token=token,lease_until=clock_timestamp()+interval '15 minutes',
attempts=a.attempts+1 FROM picked WHERE a.alert_key=picked.alert_key RETURNING a.*;
END $$;
CREATE OR REPLACE FUNCTION hfa.begin_send(k text,token text) RETURNS text LANGUAGE plpgsql AS $$
BEGIN
UPDATE hfa.alerts SET state='sending',lease_until=clock_timestamp()+interval '15 minutes'
WHERE alert_key=k AND state='leased' AND lease_token=token AND lease_until>clock_timestamp();
IF NOT FOUND THEN RAISE EXCEPTION 'Alert lease expired or was superseded'; END IF;
RETURN k;
END $$;
CREATE OR REPLACE FUNCTION hfa.finish_send(k text,token text,response jsonb) RETURNS text LANGUAGE plpgsql AS $$
DECLARE outcome text;
BEGIN
outcome := CASE WHEN response->'ok'='true'::jsonb AND NULLIF(response->>'ts','') IS NOT NULL
AND NULLIF(response->>'channel','') IS NOT NULL THEN 'delivered'
WHEN response->'ok'='false'::jsonb AND response->>'error' IN
('channel_not_found','not_in_channel','invalid_auth','missing_scope','no_text','msg_too_long') THEN 'failed'
ELSE 'uncertain' END;
UPDATE hfa.alerts SET state=outcome,slack_ts=response->>'ts',slack_channel=response->>'channel',
last_error=CASE WHEN outcome='delivered' THEN NULL ELSE coalesce(response->>'error','Unrecognized Slack response') END
WHERE alert_key=k AND state IN ('sending','uncertain') AND lease_token=token;
IF NOT FOUND THEN RAISE EXCEPTION 'No matching send attempt'; END IF;
RETURN outcome;
END $$;
COMMIT;
