UN-3513 [FEAT] Type chord-callback boundary with BatchExecutionResult / FileExecutionResult

[muhammad-ali-e]

What

• Extend two dataclasses in unstract/core/src/unstract/core/worker_models.py (strictly additive — every new field is optional with a safe default):
  • BatchExecutionResult gains skipped_already_completed: int = 0, skipped_active_duplicate: int = 0, organization_id: str | None = None.
  • FileExecutionResult gains file_name: str | None = None, result_data: Any | None = None, skipped: str | None = None.
  • Both from_dict updated to populate the new fields.
• workers/file_processing/tasks.py producers now build the typed dataclasses and emit .to_dict() instead of hand-rolled dicts:
  • L901 (general path, process_file_batch) → BatchExecutionResult(...).to_dict().
  • L1706 / L1798 / L1823 (API path, _process_file_batch_api_core helpers) → FileExecutionResult(...).to_dict(). L1798 preserves the legacy storage_result field via dict-spread merge.
• New workers/tests/test_chord_callback_boundary.py — 14 tests, 3 classes (wire-shape characterisation for both dataclasses + consumer-tolerance check).

Why

Phase 5 of the rollout asks for every queue-crossing payload to be machine-readable so a future scheduler has structured data to dispatch on. The chord-callback boundary was the last loose dict-shaped handoff in workers — typed dataclasses already existed in unstract.core.worker_models but producers were returning ad-hoc dicts. This PR closes that gap on the producer side. Consumer side is already tolerant (.get(..., default) everywhere in aggregate_file_batch_results), so no consumer change is needed.

Builds on UN-3501 (#2003 — fairness key) and runs in parallel with UN-3508 (#2009 — fairness through ExecutionDispatcher). Unblocks Phase 6 (Barrier interface) which needs a typed result contract to declare on Barrier.coordinate(...).

How

• The dataclass extensions are strictly additive — every new field is optional with a default. Existing producers/consumers that don't know about the new fields are unaffected.
• The producer migration is mechanical: return {dict} → return Class(...).to_dict(). Wire shape gains a small number of default values (empty list, None, 0) that consumers ignore via .get().
• One deliberate domain-vocabulary correction on the API path: status string changes from lowercase "completed" / "failed" (ad-hoc, matched no canonical enum) → "Success" / "Failed" (the actual ApiDeploymentResultStatus per-file vocabulary). See "Can this PR break any existing features" below.

Can this PR break any existing features. If yes, please list possible items. If no, please explain why. (PS: Admins do not merge the PR without this section filled)

No on consumer code; no on backend; yes on log/observability tooling that pattern-matches the lowercase API-path status strings (deliberate, see below).

• Consumer side: aggregate_file_batch_results (shared/processing/files/time_utils.py:130) reads every field via .get(..., default). Adding fields to the wire dict is silently ignored; missing fields fall back to defaults. New tests assert this tolerance holds for the new wire shape.
• Backend: doesn't import BatchExecutionResult. Has its own unrelated FileExecutionResult in workflow_manager/endpoint_v2/dto.py:123 that uses **data construction — untouched.
• Status-vocabulary shift on the API path: the producer at L1706 / L1798 / L1823 previously returned status="completed" / "failed" — lowercase strings matching neither ExecutionStatus (uppercase, workflow-level) nor ApiDeploymentResultStatus (Success/Failed, the canonical per-file vocabulary). The lowercase strings were a pre-existing ad-hoc third variant. After this PR they emit "Success" / "Failed" — the correct per-file vocabulary.
  • Python equality consumers: none found. Grep across workers/ and backend/ for status == "completed" / status == "failed" against these chord-callback dicts is clean (matches only unrelated cleanup-result paths).
  • Customer-facing API responses: unchanged — api-deployment/tasks.py:538-542 already uses ApiDeploymentResultStatus.SUCCESS.value / FAILED.value when building the API response. The lowercase strings were internal chord-callback noise, never the external contract.
  • Observability: dashboards / log queries / runbooks that grep worker logs for the lowercase strings need to be updated. Calling this out explicitly so ops can audit before merge.

Database Migrations

None.

Env Config

None.

Relevant Docs

None — dataclass docstrings document the new optional fields inline.

Related Issues or PRs

• Ticket: UN-3513 (sub-task) under UN-3500 (Story: PG Queue Phase 5 — Payload typing) under UN-3445 (Epic).
• Sibling PR currently in review: UN-3508 (UN-3508 [FEAT] Plumb fairness through ExecutionDispatcher #2009 — fairness through ExecutionDispatcher). No file overlap with this PR.
• Unblocks: UN-3504 (Phase 6 — Barrier interface needs a typed result contract).

Dependencies Versions

None.

Notes on Testing

cd workers
.venv/bin/python -m pytest tests/test_chord_callback_boundary.py
# 14 passed

cd unstract/sdk1
uv run pytest tests/test_execution.py
# 80 passed (existing worker_models round-trip tests confirm the additive extensions don't break anything)

Dev-tested locally against a running worker stack — workflow execution completes end-to-end; callback aggregates correctly; API-path responses unchanged.

Screenshots

N/A (no UI surface).

Checklist

I have read and understood the Contribution Guidelines.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 385c1058-12b2-46f0-8dd4-ca606bb6c487

📥 Commits

Reviewing files that changed from the base of the PR and between 8559e39 and fb3500e.

📒 Files selected for processing (3)

• unstract/core/src/unstract/core/worker_models.py
• workers/file_processing/tasks.py
• workers/tests/test_chord_callback_boundary.py

🚧 Files skipped from review as they are similar to previous changes (1)

• workers/file_processing/tasks.py

---

Summary by CodeRabbit

Release Notes

• New Features
  • Added skip reason tracking for files (e.g., already completed, active duplicate)
  • Enhanced batch processing responses with detailed execution metrics and file-level results
  • Batch results now include skip counters for visibility into skipped file breakdown
  • Standardized result format across file processing API endpoints

Walkthrough

The PR introduces typed result models (SkipReason, FileExecutionResult, BatchExecutionResult) to standardize the worker execution wire contract. Skip reasons are tracked per-file and aggregated at batch level; legacy alias fields enable backwards-compatible serialization. Producers now emit typed dicts instead of raw dictionaries. A comprehensive test suite validates round-trips and contract boundaries.

Changes

Skip-reason vocabulary and typed result models

Layer / File(s)	Summary
Skip-reason vocabulary and FileExecutionResult extensions unstract/core/src/unstract/core/worker_models.py	SkipReason enum defines ALREADY_COMPLETED and ACTIVE_DUPLICATE values. FileExecutionResult gains legacy alias fields (file_name, result_data, skipped, storage_result) with defaults; _parse_skipped() leniently converts wire values to typed enum or None with warning; from_dict() populates aliases and derives typed skipped via _parse_skipped().
BatchExecutionResult extensions and counter derivation unstract/core/src/unstract/core/worker_models.py	BatchExecutionResult adds skipped-file sub-counters (skipped_already_completed, skipped_active_duplicate) and organization_id; to_dict() strips None values from nested file_results; from_dict() populates new fields with defaults; from_file_results() derives success/failed/skipped totals from typed FileExecutionResult items.
Public export of SkipReason unstract/core/src/unstract/core/__init__.py	Re-exports SkipReason in import and __all__ list.
Producer: single-file API result standardization workers/file_processing/tasks.py	Expands imports to include typed models; _process_single_file_api() emits FileExecutionResult(...).to_dict() for all branches: already-completed with SUCCESS + ALREADY_COMPLETED, success with SUCCESS + result_data/storage_result, failure with FAILED + error.
Producer: batch aggregation and timing workers/file_processing/tasks.py	process_file_batch_api() tracks batch_start_time, computes execution_time, converts per-file results via FileExecutionResult.from_dict(), derives skipped_already_completed. _compile_batch_result() returns BatchExecutionResult(...).to_dict() with counts, timing, and organization scope.
Consumer contract tests and producer validation workers/tests/test_chord_callback_boundary.py	New test module with TestBatchExecutionResultWireShape, TestFileExecutionResultWireShape, TestProducerBinding, and TestRealConsumerTolerance classes. Validates serialization round-trips, lenient parsing of unknown skip values, producer wire shape emission, and real consumer tolerance for typed dicts.

🎯 3 (Moderate) | ⏱️ ~25 minutes

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 48.89% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and specifically summarizes the main change: typing the chord-callback boundary with BatchExecutionResult and FileExecutionResult dataclasses.
Description check	✅ Passed	The description is comprehensive and well-structured, covering all required template sections including What, Why, How, breaking changes assessment, testing notes, and checklist acknowledgment.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch UN-3513-chord-callback-result-typing

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[muhammad-ali-e]

PR Review Toolkit — multi-agent review of UN-3513 (devil's-advocate stance)

Ran 6 agents (code-reviewer, silent-failure-hunter, type-design-analyzer, pr-test-analyzer, comment-analyzer, code-simplifier) against the single commit d5a6e87f. Findings cross-validated against the actual source — links verified, not just agent assertions.

Top-line

• The dataclass extensions are mostly additive as claimed, and consumer tolerance via .get() largely holds. The PR is mergeable in spirit.
• But one BLOCKER reappeared in 5/6 agents independently: __post_init__ silently overwrites the constructor-passed status argument. Every migrated call site looks like it's setting status; none of them are. Today it's accidentally correct because each site happens to match the error-presence rule, but the contract is a lie. from_dict has the same pathology one layer deeper — it discards the wire status and re-derives.
• One HIGH that contradicts the PR's stated scope: _process_file_batch_api at tasks.py:1659 still returns a hand-rolled {successful_files, failed_files} 2-key dict — the actual API batch boundary feeding process_batch_callback_api. The PR description says "L1706/L1798/L1823 (API path returns from _process_file_batch_api_core helpers)" but those are per-file helpers, not the batch return. The boundary is half-typed.
• One HIGH that's subtle: serialize_dataclass_to_dict drops None-valued keys (data_models.py:333). That means the new alias fields (file_name, result_data, skipped) DISAPPEAR from the wire whenever a producer doesn't set them. The test guarantees about aliases hold only because the tests explicitly populate both sides; any future call site that omits the alias silently changes the wire-shape contract for downstream consumers doing if "file_name" in wire membership checks.
• Other findings cluster on: missing producer-side tests, hollow consumer-tolerance test, pre-existing dead status == "error" branch in time_utils.py:181 (PR neither breaks nor fixes), no deprecation marker on aliases, no invariant on total_files == sum(...).

Prioritized summary

#	Severity	File:line	Issue
1	BLOCKER	worker_models.py:268	__post_init__ ignores status= arg — silently coerces
2	BLOCKER	worker_models.py:296-301 (via :313)	from_dict discards wire status, recomputes from error
3	HIGH	tasks.py:1659 (anchored at :913)	API batch return still hand-rolled 2-key dict — boundary half-typed
4	HIGH	worker_models.py:264	serialize_dataclass_to_dict strips Nones — aliases drop off wire silently
5	HIGH	tasks.py:1806	Dict-spread merge can wrap storage soft-failures as SUCCESS
6	HIGH	tasks.py:1709	Already-completed skip indistinguishable from SUCCESS at aggregation
7	MEDIUM	tasks.py:1839 (time_utils.py:181)	Pre-existing dead status == "error" consumer branch
8	MEDIUM	test_chord_callback_boundary.py:25	No producer-side tests; legacy-dict regression at producer is invisible
9	MEDIUM	test_chord_callback_boundary.py:184	Hollow consumer-tolerance test re-implements .get(); doesn't drive real consumer
10	LOW	worker_models.py:246	No deprecation marker / migration plan on aliases — becomes permanent
11	LOW	worker_models.py:347	No invariant total_files == sum(...); organization_id is context, not outcome
12	NIT	test_chord_callback_boundary.py:207	Flag pre-existing skipped_files read/no-write asymmetry

Inline comments below carry the per-site detail and suggested fix.

[muhammad-ali-e]

Review feedback addressed — 8e16eb60

12/12 toolkit findings + 1 SonarCloud responded to. Triage summary:

Fixed in this commit (7)

#	Finding	What changed
4 (LOW)	Aliases lack sunset plan	TODO(UN-3516) markers on the three alias fields. UN-3516 filed with full audit/migration/removal acceptance criteria.
6 (HIGH)	L1659 not migrated — half-typed boundary	Migrated to BatchExecutionResult.to_dict(). file_results, total_files, skipped_already_completed, organization_id now on the API-path wire.
10 (MEDIUM)	Tests don't drive real producers	New TestProducerBinding drives _compile_batch_result + _process_single_file_api directly. A revert at the producer site now fails loud.
11 (MEDIUM)	Hollow consumer-tolerance test	Replaced with TestRealConsumerTolerance that imports the real aggregate_file_batch_results.
12 (NIT)	skipped_files == 0 reads as endorsement	Assertion removed.
— (SonarCloud python:S1244)	Float equality in test	pytest.approx.
— (your typo-safety question)	"already_completed" as bare string	New SkipReason StrEnum with ALREADY_COMPLETED and ACTIVE_DUPLICATE (mirroring the batch-counter vocab on BatchExecutionResult). Producer + tests use the enum.

Partially fixed (2)

#	Finding	What changed
3 (HIGH)	serialize_dataclass_to_dict strips None	Touching shared infra is out of scope; added test_none_valued_optional_fields_stripped_from_wire to document and assert the strip behaviour. Membership-check consumers now get a loud test failure if the wire shape ever diverges.
7 (HIGH)	Skipped invisible + reprocessing on except	Skipped-invisible fixed via #6. Reprocessing on except Exception at L1719 is a pre-existing latent bug — deferred.

Pushed back / deferred (5) — separate tickets to follow

#	Finding	Why deferred
1 (BLOCKER)	__post_init__ silently clobbers status	Pre-existing dataclass design; touches a class shared with backend. Own design discussion.
2 (BLOCKER)	from_dict discards wire status	Same pathology as #1; bundled into the same ticket.
5 (MEDIUM)	No total_files invariant + organization_id data-model question + add_file_result not updated	Strict invariant requires separating successful/skipped counter semantic — that's a behavioural regression for the API path's existing counter. Bundled as a follow-up.
8 (HIGH)	Storage soft-failure → SUCCESS	Pre-existing pattern (legacy dict had the same behaviour). This PR preserves it via dict-spread — does not make it worse. Separate ticket.
9 (MEDIUM)	Dead aggregator branch (status == "error")	Pre-existing dead code surfaced by the vocab shift, not caused by it. Bundled with the other latent-bug ticket.

Tests

• tests/test_chord_callback_boundary.py: 14 → 18 (added TestProducerBinding, the None-strip doc test, test_active_duplicate_skip_reason_round_trips, and TestRealConsumerTolerance's second multi-batch case)
• sdk1 tests/test_execution.py: 80/80 still green
• SonarCloud float-equality finding: resolved

Branch state

UN-3513-chord-callback-result-typing at 8e16eb60e16e2dd55f49163678267b20eb7c2467. Two commits on the branch: d5a6e87f (initial) + 8e16eb60 (this review-fix round). CI / SonarCloud / Greptile / CodeRabbit re-running on the new HEAD.

[muhammad-ali-e]

Second-pass review — three new findings that escaped the first round. All correctness-level: one data-loss / broken-contract at the batch boundary, one rolling-deploy regression on the consumer read path, and a test-coverage gap where the TestProducerBinding docstring overpromises relative to what the class actually exercises.

[greptile-apps]

Greptile Summary

This PR completes the chord-callback payload typing for the worker layer by replacing ad-hoc dict returns with typed dataclasses (BatchExecutionResult / FileExecutionResult) at every producer site. All new fields are optional with safe defaults, so the change is wire-additive and requires no consumer changes.

• worker_models.py: Adds SkipReason enum, extends both dataclasses with optional fields, fixes BatchExecutionResult.to_dict() to strip None from nested file-result dicts so standalone and batch-embedded wire shapes are symmetric, and adds a from_file_results factory method.
• tasks.py: Migrates three producer sites from hand-rolled dicts to FileExecutionResult(...).to_dict() and one site to BatchExecutionResult(...).to_dict(); status vocabulary on the API path is corrected from ad-hoc lowercase strings to the canonical ApiDeploymentResultStatus values.
• test_chord_callback_boundary.py: 14 new tests across three layers (wire-shape round-trip, real producer binding, real consumer tolerance) lock the on-wire contract.

Confidence Score: 5/5

Safe to merge; all new fields are optional with safe defaults, the consumer already uses .get() everywhere, and the status-vocabulary change is intentional and well-scoped.

Every new field in both dataclasses is optional with a safe default, making the wire change additive. The None-stripping fix in BatchExecutionResult.to_dict() was the last remaining asymmetry flagged in prior review and is now covered by a dedicated test. The status-vocabulary correction is deliberate and the PR description explicitly calls out that no Python equality consumer matches the old lowercase strings. Three layers of tests (wire-shape, producer binding, consumer tolerance) provide strong regression coverage for the changed paths.

No files require special attention. The most complex change is the BatchExecutionResult.to_dict() None-stripping fixup, which is well-tested and clearly documented.

Important Files Changed

Filename	Overview
unstract/core/src/unstract/core/worker_models.py	Adds SkipReason enum, extends FileExecutionResult/BatchExecutionResult with optional fields, fixes None-stripping asymmetry in BatchExecutionResult.to_dict(), and introduces from_file_results factory; all changes are strictly additive and well-commented.
workers/file_processing/tasks.py	Migrates three _process_single_file_api return paths and _compile_batch_result from hand-rolled dicts to typed dataclass .to_dict() calls; status vocabulary corrected to canonical ApiDeploymentResultStatus values; batch_start_time added to enable execution-time tracking on the API path.
workers/tests/test_chord_callback_boundary.py	New 771-line test file with 14 tests across three classes: wire-shape characterisation, real producer binding, and real consumer tolerance. Covers round-trips, JSON safety, None stripping symmetry, canonical vocab validation, and lenient unknown-SkipReason handling.
unstract/core/src/unstract/core/init.py	Adds SkipReason to the module's public import and all list; purely mechanical re-export change.

Sequence Diagram

sequenceDiagram
    participant GW as General Path<br/>(process_file_batch)
    participant API as API Path<br/>(process_file_batch_api)
    participant SF as _process_single_file_api
    participant CR as _compile_batch_result
    participant CB as Chord Callback<br/>(aggregate_file_batch_results)

    GW->>CR: context (WorkflowContextData)
    CR->>CB: "BatchExecutionResult(...).to_dict()<br/>{total_files, successful_files, skipped_*, org_id, ...}"

    API->>SF: file_data (per file)
    SF-->>API: "FileExecutionResult(...).to_dict()<br/>{file, file_name, status=Success/Failed, skipped?, ...}"
    API->>API: FileExecutionResult.from_dict(r) for each result
    API->>CB: "BatchExecutionResult(...).to_dict()<br/>{file_results: [stripped-None dicts], ...}"

    CB->>CB: "aggregate via .get(..., default)<br/>(consumer tolerance preserved)"

Loading

_{Reviews (4): Last reviewed commit: "UN-3513 [FIX] Address vishnuszipstack re..." | Re-trigger Greptile}

[vishnuszipstack]

Automated PR Review — PR Review Toolkit (6 agents)

Ran Comment Analyzer, PR Test Analyzer, Silent Failure Hunter, Type Design Analyzer, Code Reviewer, and Code Simplifier over the chord-callback typing change.

Headline: Code Reviewer found no correctness bugs — the to_dict/from_dict round-trips preserve storage_result/result_data/file_name/skipped, execution_time is computed on every return path of process_file_batch_api, and to_api_dict is never reached by the chord producers, so the canonical-vs-alias field split causes no live data loss. The PR is functionally safe to merge.

The inline comments below are design hardening, silent-failure robustness, and test-coverage findings — none are blocking bugs. Highest-leverage items: make status a derived property and collapse the alias pairs (removes two classes of representable-illegal-state), and add the missing _parse_skipped unknown-value test (the one documented crash-prevention path with zero coverage).

Severity legend: [Important] = should fix, [Suggestion] = nice to have.

[sonarqubecloud]

Quality Gate passed

Issues
1 New issue
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[github-actions]

Unstract test results

Per-group results

Status	Group	Tier	Passed	Failed	Errors	Skipped	Duration (s)
❌	unit-connectors	unit	64	12	0	3	16.6
❌	unit-core	unit	0	0	2	0	1.2
❌	unit-platform-service	unit	9	0	1	0	1.3
✅	unit-prompt-service	unit	15	0	0	0	19.5
✅	unit-rig	unit	53	0	0	0	3.3
✅	unit-runner	unit	11	0	0	0	3.1
✅	unit-sdk1	unit	381	0	0	0	23.0
❌	unit-tool-registry	unit	0	0	1	0	1.3
⚪	unit-workers	unit	0	0	0	0	17.3
	TOTAL		533	12	4	3	86.7

Critical paths

⚠️ Critical paths not yet covered

• auth-login — User can log in and obtain a session cookie. (entry: POST /api/v1/auth/login; declared coverage: no groups declared)
• adapter-register-llm — Register and validate an LLM adapter. (entry: POST /api/v1/adapter/; declared coverage: no groups declared)
• workflow-create-execute — Create a workflow, configure source+destination, execute, poll, fetch result. (entry: POST /api/v1/workflow/{id}/execute/; declared coverage: e2e-workflow)
• api-deployment-run — Deploy a workflow as an API, POST a document, receive structured JSON. (entry: POST /deployment/api/{org}/{name}/; declared coverage: e2e-api-deployment)
• prompt-studio-fetch-response — Prompt Studio: create project, add prompt, run single-pass, get response. (entry: POST /api/v1/prompt-studio/prompt-studio-tool/{id}/fetch_response/; declared coverage: e2e-prompt-studio)
• pipeline-etl-execute — Run an ETL pipeline from source connector to destination. (entry: POST /api/v1/pipeline/{id}/execute/; declared coverage: no groups declared)
• usage-token-tracking — Per-execution token usage is recorded and retrievable. (entry: GET /api/v1/usage/get_token_usage/; declared coverage: no groups declared)
• workflow-execution-fan-out — Multi-file workflow execution fans out to file-processing workers and rejoins. (entry: internal: backend → rabbitmq → workers/file_processing; declared coverage: no groups declared)
• callback-result-delivery — Async results are posted back via the callback worker. (entry: internal: workers/callback → backend /internal endpoints; declared coverage: no groups declared)

✅ Covered critical paths

• tool-sandbox-exec — covered by unit-runner