1"""The shared Actor pipeline every listing calls (SPEC v2 §4, §5.12, §8, §9.1, §13.4, §14).
2
3Every Actor under ``actors/`` is a ~30-line ``src/main.py`` that calls :func:`main` here.
4The per-ATS listings (A2–A4, §3.2) pass ``provider=`` to pin themselves to one ATS: the
5provider selector disappears from their input schema, a bare token in ``companies``
6becomes that provider's own board slug, and an entry naming a *different* ATS becomes a
7free error row instead of a silent cross-provider fetch. Nothing else differs — same
8adapters, same normalization, same `job` / `delta-run` events, same price.
9
10Pipeline, in order, and everything before the last step is free:
11
12 input (+ key aliases, §4.1) -> resolve (§5.11) -> adapter fetch (§5) -> description /
13 redaction / raw post-processing (§4.1) -> filters (§4.1) -> dedupe (§4.5.6) ->
14 onlyNewJobs delta (§4.5.6) -> push, charging `job` once per delivered row (§8.2) ->
15 free company_summary and error rows (§4.6)
16
17The rules this file exists to enforce (§8.2, §5.12, §10.1 `test_billing.py`):
18
19* `job` is the only paid event, charged once per delivered row — `core.billing` owns
20 every `push_data` call so that stays true.
21* Company summaries, error rows, filtered-out jobs, failed companies and companies never
22 reached because `maxJobs` or the charge limit fired are all free.
23* `delta-run` is charged once per run when `onlyNewJobs` is on, never otherwise.
24* One bad company never ends the run: it becomes a typed `error` row (§13.4 #14).
25"""
26
27from __future__ import annotations
28
29import asyncio
30import inspect
31import time
32from collections import Counter
33from dataclasses import dataclass, field
34from datetime import UTC, datetime
35from types import ModuleType
36from typing import Any
37
38from apify import Actor, Event
39
40from core.billing import Billing
41from core.directory import get_directory
42from core.filters import Filters
43from core.http import FetchError, make_client
44from core.models import PROVIDERS, Ref
45from core.normalize.redact import redact_description
46from core.providers import AdapterNotFound, get_adapter
47from core.resolve import Unresolved, needs_directory, resolve
48from core.state import SeenState
49
50
51COMPANY_BUDGET_SECS = 120.0
52
53
54INPUT_ALIASES = (
55 "queries",
56 "companyUrls",
57 "startUrls",
58 "boardTokens",
59 "siteNames",
60 "jobBoardNames",
61 "subdomains",
62 "companyIdentifiers",
63)
64
65
66
67DEFAULTS: dict[str, Any] = {
68 "providers": list(PROVIDERS),
69 "maxJobs": 1000,
70 "maxJobsPerCompany": 0,
71 "titleKeywords": [],
72 "excludeTitleKeywords": [],
73 "locationKeywords": [],
74 "remoteOnly": False,
75 "departments": [],
76 "employmentTypes": [],
77 "strictEmploymentType": False,
78 "postedAfter": None,
79 "includeDescription": True,
80 "descriptionFormat": "text",
81 "redactContacts": True,
82 "outputProfile": "full",
83 "includeCompanySummary": True,
84 "includeRawJson": False,
85 "dedupe": "id",
86 "onlyNewJobs": False,
87 "stateKey": "ats-jobs-state-default",
88 "stateRetentionDays": 90,
89 "maxConcurrency": 8,
90 "requestTimeoutSecs": 30,
91 "failOnAllErrors": False,
92}
93
94
95COMPANY_STATE_KEY = "companies"
96
97
98
99
100DEGRADED_RATIO = 0.9
101DEGRADED_MIN_COMPANIES = 5
102
103
104MAX_COMPANIES = 2000
105
106
107
108BUDGET_HEADROOM_SECS = 10.0
109
110
111
112
113
114def now_iso() -> str:
115 return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131NUMERIC_BOUNDS: dict[str, tuple[int, int]] = {
132 "maxJobs": (0, 200_000),
133 "maxJobsPerCompany": (0, 20_000),
134 "maxConcurrency": (1, 32),
135 "requestTimeoutSecs": (5, 120),
136 "stateRetentionDays": (0, 730),
137}
138
139
140def _bounded(key: str, value: Any) -> int:
141 """One numeric knob, coerced into its schema bounds. Never raises."""
142 low, high = NUMERIC_BOUNDS[key]
143 try:
144 number = int(float(value))
145 except (TypeError, ValueError, OverflowError):
146
147
148
149 number = int(DEFAULTS[key])
150 return min(high, max(low, number))
151
152
153def read_config(raw: dict[str, Any], provider: str | None = None) -> dict[str, Any]:
154 """Apply defaults without letting a meaningful falsy value (`maxJobs: 0`) be lost.
155
156 ``provider`` is the §3.2 pin. It overrides `providers` outright — the per-ATS input
157 schemas have no selector, so anything arriving under that key came from an API caller
158 replaying a multi-ATS input, and honouring it would deliver another ATS's jobs from a
159 listing that names one. The delta store is namespaced with it too: the named KV store
160 is account-wide, so a shared default would have the Greenhouse listing marking ids
161 that then went missing from the multi-ATS listing's own delta.
162 """
163 cfg = {
164 key: (raw[key] if raw.get(key) is not None else value) for key, value in DEFAULTS.items()
165 }
166 if provider:
167 cfg["providers"] = [provider]
168 if raw.get("stateKey") is None:
169 cfg["stateKey"] = f"{provider}-jobs-state-default"
170 else:
171 cfg["providers"] = [p for p in cfg["providers"] if p in PROVIDERS] or list(PROVIDERS)
172 for key in NUMERIC_BOUNDS:
173 cfg[key] = _bounded(key, cfg[key])
174 return cfg
175
176
177def read_companies(raw: dict[str, Any]) -> list[str]:
178 """`companies`, or the first non-empty §4.1 alias. Order preserved, repeats dropped."""
179 values = raw.get("companies") or next(
180 (raw[alias] for alias in INPUT_ALIASES if raw.get(alias)), []
181 )
182
183
184
185 if not isinstance(values, list):
186 values = [values] if isinstance(values, str) else []
187 out: list[str] = []
188 for value in values:
189 if isinstance(value, dict):
190 value = value.get("url") or value.get("value") or value.get("slug")
191 if isinstance(value, str) and value.strip():
192 out.append(value.strip())
193
194
195 return list(dict.fromkeys(out))[:MAX_COMPANIES]
196
197
198
199
200
201def error_item(
202 provider: str | None, slug: str | None, entry: str | None, status: str, message: str
203) -> dict[str, Any]:
204 """A free `error` row (§4.6). Typed status, human message — never a stack trace."""
205 return {
206 "recordType": "error",
207 "provider": provider,
208 "companySlug": slug,
209 "status": status,
210 "error": message,
211 "scrapedAt": now_iso(),
212 "input": entry,
213 }
214
215
216def summary_item(
217 ref: Ref,
218 *,
219 status: str,
220 company: str | None = None,
221 domain: str | None = None,
222 jobs_found: int | None = None,
223 jobs_kept: int | None = None,
224 new_jobs: int | None = None,
225 duplicates: int | None = None,
226 tracked_since: str | None = None,
227 top_departments: list[dict[str, Any]] | None = None,
228 warnings: list[str] | None = None,
229) -> dict[str, Any]:
230 """A free `company_summary` row (§4.6). Never charged, ever."""
231 return {
232 "recordType": "company_summary",
233 "provider": ref.provider,
234 "companySlug": ref.slug,
235 "company": company,
236 "companyDomain": domain,
237 "status": status,
238 "jobsFound": jobs_found,
239 "jobsKept": jobs_kept,
240 "newJobs": new_jobs,
241 "duplicatesDropped": duplicates,
242 "trackedSince": tracked_since,
243 "topDepartments": top_departments,
244 "scrapedAt": now_iso(),
245 "input": ref.input,
246 "warnings": warnings or None,
247 }
248
249
250def top_departments(records: list[Any], limit: int = 3) -> list[dict[str, Any]] | None:
251 counts = Counter(r.department for r in records if r.department)
252 return [{"name": name, "count": count} for name, count in counts.most_common(limit)] or None
253
254
255def classify(exc: BaseException) -> tuple[str, str]:
256 """Map an adapter failure onto a §5.12 status plus a message a buyer can act on."""
257 if isinstance(exc, FetchError):
258 return exc.status, str(exc)
259 if isinstance(exc, TimeoutError):
260 return "timeout", f"the company took longer than {COMPANY_BUDGET_SECS:.0f}s"
261 if isinstance(exc, ValueError | KeyError | TypeError):
262 return "parse_error", f"the provider payload did not parse: {type(exc).__name__}"
263
264
265 return "http_error", f"the company could not be fetched ({type(exc).__name__})"
266
267
268
269
270
271def shape_record(record: Any, cfg: dict[str, Any]) -> None:
272 """Apply the §4.1 output switches: description, format, redaction, raw payload.
273
274 Redaction is re-applied here rather than trusted to the adapter: it is the mechanism
275 behind the §15.2 no-PII claim, and `redact_text` on already-clean text is a no-op.
276 """
277 if not cfg["includeDescription"]:
278 record.descriptionHtml = None
279 record.descriptionText = None
280 record.descriptionRedacted = None
281 else:
282 if cfg["descriptionFormat"] == "text":
283 record.descriptionHtml = None
284 elif cfg["descriptionFormat"] == "html":
285 record.descriptionText = None
286 if record.descriptionHtml or record.descriptionText:
287 html, text, redacted = redact_description(
288 record.descriptionHtml, record.descriptionText, cfg["redactContacts"]
289 )
290 record.descriptionHtml, record.descriptionText = html, text
291 if redacted is not None:
292 record.descriptionRedacted = bool(record.descriptionRedacted) or redacted
293 if not cfg["includeRawJson"]:
294 record.raw = None
295
296
297async def adapter_fetch(module: ModuleType, ref: Ref, client: Any, cfg: dict[str, Any]) -> list:
298 """Call `fetch(ref, client)`, passing the input dict too when the adapter takes it."""
299 params = inspect.signature(module.fetch).parameters
300 if "options" in params:
301 result = module.fetch(ref, client, options=cfg)
302 elif len(params) >= 3:
303 result = module.fetch(ref, client, cfg)
304 else:
305 result = module.fetch(ref, client)
306 return list(await result) if inspect.isawaitable(result) else list(result)
307
308
309
310
311
312@dataclass(slots=True)
313class RunCtx:
314 cfg: dict[str, Any]
315 filters: Filters
316 billing: Billing
317
318 provider: str | None = None
319 state: SeenState | None = None
320 company_state: dict[str, dict[str, Any]] = field(default_factory=dict)
321 lock: asyncio.Lock = field(default_factory=asyncio.Lock)
322 seen_ids: set[str] = field(default_factory=set)
323
324
325
326
327
328
329
330
331
332 content_survivors: dict[str, list[str]] = field(default_factory=dict)
333 summaries: list[tuple[str, dict[str, Any]]] = field(default_factory=list)
334 companies_seen: Counter = field(default_factory=Counter)
335 companies_zero: Counter = field(default_factory=Counter)
336 jobs_found_total: int = 0
337 companies_ok: int = 0
338 store: Any = None
339
340 flushed: bool = False
341
342
343def company_key(ref: Ref) -> str:
344 """Lower-cased for lookup only — a fetch URL is never rebuilt from it (§5.11)."""
345 return f"{ref.provider}:{ref.slug.casefold()}"
346
347
348
349
350
351def dedupe(ctx: RunCtx, records: list[Any]) -> tuple[list[Any], int]:
352 """§4.5.6: always by `id`, additionally by `contentKey` when asked. Run-wide, and the
353 surviving row carries the ids it swallowed in `dedupedFrom`."""
354 kept: list[Any] = []
355 dropped = 0
356 for record in records:
357 if record.id:
358 if record.id in ctx.seen_ids:
359 dropped += 1
360 continue
361 ctx.seen_ids.add(record.id)
362 if ctx.cfg["dedupe"] == "content" and record.contentKey:
363 survivor = ctx.content_survivors.get(record.contentKey)
364 if survivor is not None:
365
366
367
368 survivor.append(record.id or "")
369 dropped += 1
370 continue
371
372
373 record.dedupedFrom = record.dedupedFrom or []
374 ctx.content_survivors[record.contentKey] = record.dedupedFrom
375 kept.append(record)
376 return kept, dropped
377
378
379def apply_delta(ctx: RunCtx, records: list[Any]) -> tuple[list[Any], int | None]:
380 """§4.5.6 across runs. Without a state store `isNew` stays null — we do not know.
381
382 Decides only. Marking here used to happen for every kept row — before the
383 `maxJobsPerCompany` trim, before `maxJobs` and before the charge limit could stop the
384 push — so a row that was never delivered still had `isNew=false` on every later run
385 and was lost permanently (V1 B3). :func:`commit_delta` marks what actually landed.
386 """
387 if ctx.state is None:
388 return records, None
389 for record in records:
390 if not record.id:
391 continue
392 record.isNew = ctx.state.is_new(record.id)
393 record.firstSeenAt = ctx.state.first_seen(record.id) or record.scrapedAt
394 return [r for r in records if r.isNew], 0
395
396
397def commit_delta(ctx: RunCtx, delivered: list[Any]) -> int:
398 """Mark the rows that were actually pushed. Never call it before the push (V1 B3)."""
399 if ctx.state is None:
400 return 0
401 marked = 0
402 for record in delivered:
403 if record.id and ctx.state.mark(record.id, record.changeHash):
404 marked += 1
405 return marked
406
407
408async def push_jobs(ctx: RunCtx, records: list[Any]) -> int:
409 """One `job` event per delivered row; `core.billing` decides when to stop (§8.2)."""
410 profile = ctx.cfg["outputProfile"]
411 delivered = 0
412 async with ctx.lock:
413 for record in records:
414 if not await ctx.billing.push_job(record.to_item(profile)):
415 break
416 delivered += 1
417 return delivered
418
419
420async def process_company(ref: Ref, client: Any, ctx: RunCtx) -> None:
421 cfg = ctx.cfg
422 started = time.monotonic()
423
424
425
426 budget = dict(cfg, deadline=started + COMPANY_BUDGET_SECS - BUDGET_HEADROOM_SECS)
427 try:
428 module = get_adapter(ref.provider)
429 async with asyncio.timeout(COMPANY_BUDGET_SECS):
430 records = await adapter_fetch(module, ref, client, budget)
431 except AdapterNotFound as exc:
432 await ctx.billing.push_free(
433 error_item(ref.provider, ref.slug, ref.input, "provider_unavailable", str(exc))
434 )
435 Actor.log.warning("adapter missing", extra={"provider": ref.provider, "slug": ref.slug})
436 return
437 except Exception as exc:
438 status, message = classify(exc)
439 await ctx.billing.push_free(error_item(ref.provider, ref.slug, ref.input, status, message))
440 Actor.log.warning(
441 "company failed",
442 extra={
443 "provider": ref.provider,
444 "slug": ref.slug,
445 "status": status,
446 "error": f"{type(exc).__name__}: {exc}",
447 },
448 )
449 return
450
451 ctx.companies_seen[ref.provider] += 1
452 ctx.companies_ok += 1
453 jobs_found = len(records)
454 ctx.jobs_found_total += jobs_found
455 if jobs_found == 0:
456 ctx.companies_zero[ref.provider] += 1
457
458 company = next((r.company for r in records if r.company), None)
459 domain = next((r.companyDomain for r in records if r.companyDomain), None) or ref.domain
460 scraped = now_iso()
461
462 survivors: list[Any] = []
463 for record in records:
464 record.provider = record.provider or ref.provider
465 record.companySlug = record.companySlug or ref.slug
466 record.input = record.input or ref.input
467 record.scrapedAt = record.scrapedAt or scraped
468
469
470 record.companyDomain = record.companyDomain or ref.domain
471 shape_record(record, cfg)
472 if ctx.filters.keep(record):
473 survivors.append(record)
474
475 async with ctx.lock:
476 kept, duplicates = dedupe(ctx, survivors)
477 emit, new_jobs = apply_delta(ctx, kept)
478 if cfg["maxJobsPerCompany"]:
479 emit = emit[: cfg["maxJobsPerCompany"]]
480
481 try:
482 delivered = await push_jobs(ctx, emit)
483 except Exception as exc:
484 status, message = classify(exc)
485 await ctx.billing.push_free(error_item(ref.provider, ref.slug, ref.input, status, message))
486 Actor.log.warning(
487 "push failed",
488 extra={"provider": ref.provider, "slug": ref.slug, "error": f"{type(exc)}: {exc}"},
489 )
490 return
491 if ctx.state is not None:
492 async with ctx.lock:
493
494 new_jobs = commit_delta(ctx, emit[:delivered])
495
496 warnings = list(ctx.filters.warnings)
497 tracked_since = None
498 if ctx.state is not None:
499 key = company_key(ref)
500 previous = ctx.company_state.get(key, {})
501 tracked_since = previous.get("firstSeen") or scraped[:10]
502
503
504 if jobs_found == 0 and previous.get("jobs"):
505 warnings.append("empty_suspect")
506 ctx.company_state[key] = {
507 "firstSeen": tracked_since,
508 "lastSeen": scraped[:10],
509 "jobs": jobs_found,
510 }
511
512 status = ctx.billing.stop_status if delivered < len(emit) else "ok"
513 ctx.summaries.append(
514 (
515 ref.provider,
516 summary_item(
517 ref,
518 status=status or "ok",
519 company=company,
520 domain=domain,
521 jobs_found=jobs_found,
522 jobs_kept=len(kept),
523 new_jobs=new_jobs,
524 duplicates=duplicates,
525 tracked_since=tracked_since,
526 top_departments=top_departments(kept),
527 warnings=warnings,
528 ),
529 )
530 )
531 Actor.log.info(
532 "company done",
533 extra={
534 "provider": ref.provider,
535 "slug": ref.slug,
536 "jobsFound": jobs_found,
537 "jobsKept": len(kept),
538 "jobsDelivered": delivered,
539 "newJobs": new_jobs,
540 "duplicatesDropped": duplicates,
541 "seconds": round(time.monotonic() - started, 2),
542 },
543 )
544
545
546async def worker(queue: asyncio.Queue[Ref], client: Any, ctx: RunCtx) -> None:
547 while True:
548 try:
549 ref = queue.get_nowait()
550 except asyncio.QueueEmpty:
551 return
552 try:
553 if ctx.billing.stopped:
554
555 ctx.summaries.append(
556 (ref.provider, summary_item(ref, status=ctx.billing.stop_status or "ok"))
557 )
558 continue
559 await process_company(ref, client, ctx)
560 finally:
561 queue.task_done()
562
563
564async def resolve_all(entries: list[str], client: Any, ctx: RunCtx) -> list[Ref]:
565 """§5.11. The directory is loaded only if some entry actually needs it (§6.6).
566
567 ``client`` is not optional: `load_directory` only offers the jsDelivr and
568 raw.githubusercontent sources when it has one, so calling `get_directory()` bare left
569 the loader with nothing but an empty KV store and a baked file that is not in the tree
570 — every bare slug and company name became an error row (V1 H2, V3 S10).
571 """
572 directory = None
573 if any(needs_directory(entry, ctx.provider) for entry in entries):
574 try:
575 directory = await get_directory(client)
576 except Exception as exc:
577 Actor.log.warning("company directory unavailable", extra={"error": str(exc)})
578
579 refs: list[Ref] = []
580 for entry in entries:
581
582
583
584
585 result = resolve(
586 entry, providers=ctx.cfg["providers"], directory=directory, pin=ctx.provider
587 )
588 if isinstance(result, Unresolved):
589 await ctx.billing.push_free(
590 error_item(None, None, entry, result.status, result.message)
591 )
592 continue
593 refs.append(result)
594 return refs
595
596
597async def flush_summaries(ctx: RunCtx) -> None:
598 """§5.12 degradation guard, then push every free summary row."""
599 degraded = {
600 provider
601 for provider, seen in ctx.companies_seen.items()
602 if seen >= DEGRADED_MIN_COMPANIES and ctx.companies_zero[provider] / seen > DEGRADED_RATIO
603 }
604 for provider in degraded:
605 Actor.log.error(
606 "provider degraded: 200-with-zero-jobs at population scale — see §14.3",
607 extra={"provider": provider, "companies": ctx.companies_seen[provider]},
608 )
609 if ctx.flushed:
610 return
611 ctx.flushed = True
612 if not ctx.cfg["includeCompanySummary"]:
613 return
614 for provider, item in ctx.summaries:
615 if provider in degraded:
616 item["warnings"] = (item.get("warnings") or []) + ["provider_degraded"]
617
618
619 if item.get("status") == "ok":
620 item["status"] = "provider_degraded"
621 await ctx.billing.push_free(item)
622
623
624async def open_state(ctx: RunCtx) -> None:
625 """`onlyNewJobs` state, degrading instead of failing the run (V1 M5).
626
627 Apify rejects a named store whose name it does not like, and that rejection used to
628 surface as an unhandled exception *after* every company had already resolved — the
629 opposite of the "degrade, never fail" posture the rest of the shell keeps.
630 """
631 key = ctx.cfg["stateKey"]
632 try:
633 ctx.state = await SeenState.open(key)
634 ctx.store = await Actor.open_key_value_store(name=key)
635 previous = await ctx.store.get_value(COMPANY_STATE_KEY)
636 except Exception as exc:
637 ctx.state = None
638 ctx.store = None
639 Actor.log.warning(
640 "state store unavailable; onlyNewJobs disabled", extra={"error": str(exc)}
641 )
642 await ctx.billing.push_free(
643 error_item(None, None, None, "http_error", f"could not open state store {key!r}")
644 )
645 return
646
647 ctx.company_state = previous if isinstance(previous, dict) else {}
648
649
650async def save_state(ctx: RunCtx) -> None:
651 if ctx.state is None or ctx.store is None:
652 return
653 pruned = ctx.state.prune(ctx.cfg["stateRetentionDays"])
654 await ctx.state.save()
655 await ctx.store.set_value(COMPANY_STATE_KEY, ctx.company_state)
656 Actor.log.info(
657 "state saved",
658 extra={"ids": len(ctx.state.seen), "pruned": pruned, "key": ctx.cfg["stateKey"]},
659 )
660
661
662async def main(provider: str | None = None) -> None:
663 """Run one Actor. ``provider`` pins it to a single ATS (§3.2); None = all six."""
664 async with Actor:
665 raw = await Actor.get_input() or {}
666 cfg = read_config(raw, provider)
667 entries = read_companies(raw)
668 ctx = RunCtx(
669 cfg=cfg,
670 filters=Filters.from_input(cfg),
671 billing=Billing(max_jobs=cfg["maxJobs"]),
672 provider=provider,
673 )
674
675 Actor.log.info(
676 "run start",
677 extra={
678 "companies": len(entries),
679 "providers": cfg["providers"],
680 "maxJobs": cfg["maxJobs"],
681 "onlyNewJobs": cfg["onlyNewJobs"],
682 "outputProfile": cfg["outputProfile"],
683 "filters": ctx.filters.active,
684 },
685 )
686
687 if not entries:
688 await ctx.billing.push_free(
689 error_item(
690 None,
691 None,
692 None,
693 "no_companies",
694 "No companies given. Add slugs or career-site URLs to the Companies field.",
695 )
696 )
697 return
698
699
700
701 async def on_abort(_data: Any = None) -> None:
702 await flush_summaries(ctx)
703 await save_state(ctx)
704
705 Actor.on(Event.ABORTING, on_abort)
706 Actor.on(Event.MIGRATING, on_abort)
707
708 async with make_client(timeout_secs=cfg["requestTimeoutSecs"]) as client:
709 if cfg["onlyNewJobs"]:
710
711
712
713
714
715 await ctx.billing.charge_delta_run()
716
717 refs = await resolve_all(entries, client, ctx)
718 if not refs:
719 Actor.log.warning("nothing resolved", extra={"entries": len(entries)})
720 if cfg["failOnAllErrors"]:
721 await Actor.fail(status_message="No company could be resolved to an ATS board")
722 return
723
724 if cfg["onlyNewJobs"]:
725 await open_state(ctx)
726 if ctx.state is None:
727
728
729
730
731
732
733
734 ctx.billing.budget_exhausted = True
735 Actor.log.error(
736 "onlyNewJobs requested but the state store is unavailable; "
737 "delivering nothing rather than re-charging the whole baseline"
738 )
739
740 queue: asyncio.Queue[Ref] = asyncio.Queue()
741 for ref in refs:
742 queue.put_nowait(ref)
743
744
745
746 outcomes = await asyncio.gather(
747 *(
748 asyncio.create_task(worker(queue, client, ctx))
749 for _ in range(min(cfg["maxConcurrency"], len(refs)))
750 ),
751 return_exceptions=True,
752 )
753 for outcome in outcomes:
754 if isinstance(outcome, BaseException):
755 Actor.log.exception("worker crashed", exc_info=outcome)
756
757 await flush_summaries(ctx)
758 await save_state(ctx)
759
760 Actor.log.info(
761 "run done",
762 extra={
763 "companies": len(refs),
764 "companiesOk": ctx.companies_ok,
765 "jobsFound": ctx.jobs_found_total,
766 "jobsPushed": ctx.billing.jobs_pushed,
767 "freeRows": ctx.billing.free_pushed,
768 "stopped": ctx.billing.stop_status,
769 },
770 )
771 await Actor.set_status_message(
772 f"{ctx.billing.jobs_pushed} job rows from {ctx.companies_ok}/{len(refs)} companies"
773 + (f" ({ctx.billing.stop_status})" if ctx.billing.stop_status else "")
774 )
775 if cfg["failOnAllErrors"] and ctx.jobs_found_total == 0:
776 await Actor.fail(status_message="Every company failed or returned no jobs")
