separate application execution boundaries - #17
Conversation
📝 WalkthroughWalkthroughThis PR introduces a JSON-round-trippable application boundary with execution contracts, runtime ports, local adapters, shared orchestration, client integrations, compatibility delegation, and tests for execution, serialization, cancellation, and import boundaries. ChangesApplication boundary and pipeline execution
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
loafer/application/local.py (1)
17-33: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winInteractive stdin default can hang headless/scheduled runs.
get_local_application()defaults toInputReviewPort()when no reviewer is supplied.loafer/scheduler.py::_run_pipeline_jobcallsget_local_application()without a reviewer, so any scheduled pipeline configured withtransform.review: truewill block oninput()waiting for a TTY that doesn't exist in a background job — no timeout, no failure, just an indefinite hang. Only the CLI (cli.py) explicitly overrides this withRichReviewPort(); any other new headlessLocalApplicationServiceconsumer inherits the same risk.Consider making the default a non-interactive port that fails fast (raises/rejects) when review is required but no reviewer was explicitly configured, reserving
InputReviewPortfor callers that know they have a TTY.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@loafer/application/local.py` around lines 17 - 33, The default reviewer in get_local_application must be non-interactive so headless callers such as scheduler._run_pipeline_job cannot block on stdin. Replace the implicit InputReviewPort fallback with a reviewer port that fails fast when review is requested without explicit configuration, while preserving explicitly supplied reviewers and leaving CLI-owned interactive configuration unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@loafer/application/__init__.py`:
- Around line 18-38: Update the public exports in the application package’s
__all__ to include the missing EventPublisher and ReviewPort runtime ports, and
import them from loafer.ports.runtime alongside CancellationPort,
CheckpointPort, and SecretResolver. Preserve all existing boundary exports.
In `@loafer/application/service.py`:
- Around line 246-283: Update stream_pipeline so generator closure via
GeneratorExit still executes _cleanup_source_connector and _cleanup_elt_staging,
rather than relying only on except Exception handlers. Preserve
_stream_prepared’s cancellation behavior and ensure all early-abandonment paths
finalize both cleanup operations before the generator closes.
In `@loafer/engine.py`:
- Around line 298-317: Restructure the cleanup around the streaming logic in the
enclosing generator so _cleanup_source_connector(state) and
_cleanup_elt_staging(state, mode) execute from a finally block, including when
iteration closes early. Preserve the existing PipelineError and
generic-exception handling, and keep _persist_cursor(state) only on the
successful non-dry-run path.
In `@loafer/graph/state.py`:
- Around line 6-8: Correct the module reference in the state module docstring to
point to loafer.contracts instead of loafer.application.contracts, leaving the
surrounding persistence-contract description unchanged.
---
Outside diff comments:
In `@loafer/application/local.py`:
- Around line 17-33: The default reviewer in get_local_application must be
non-interactive so headless callers such as scheduler._run_pipeline_job cannot
block on stdin. Replace the implicit InputReviewPort fallback with a reviewer
port that fails fast when review is requested without explicit configuration,
while preserving explicitly supplied reviewers and leaving CLI-owned interactive
configuration unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: c912d847-0dc1-4994-9314-b4d89f94fcee
📒 Files selected for processing (23)
CHANGELOG.mdCONTRIBUTING.mdPRODUCTION_READINESS.mdREADME.mdloafer/adapters/runtime.pyloafer/application/__init__.pyloafer/application/local.pyloafer/application/service.pyloafer/cli.pyloafer/connectors/registry.pyloafer/contracts.pyloafer/engine.pyloafer/graph/state.pyloafer/ports/runtime.pyloafer/runner.pyloafer/scheduler.pyloafer/transform/ai_runner.pyskills/loafer-engineering/SKILL.mdskills/loafer-engineering/references/architecture.mdtests/e2e/test_application_pipeline.pytests/unit/test_application_contracts.pytests/unit/test_application_service.pytests/unit/test_import_boundaries.py
| from loafer.ports.runtime import CancellationPort, CheckpointPort, SecretResolver | ||
|
|
||
| __all__ = [ | ||
| "BatchEnvelope", | ||
| "CancellationPort", | ||
| "Checkpoint", | ||
| "CheckpointPort", | ||
| "ConnectorCatalog", | ||
| "ExecutionPlan", | ||
| "LocalApplicationService", | ||
| "RunEvent", | ||
| "RunPipeline", | ||
| "RunRequest", | ||
| "RunResult", | ||
| "RunSnapshot", | ||
| "RunStatus", | ||
| "SecretResolver", | ||
| "StageStatus", | ||
| "ValidationResult", | ||
| "get_local_application", | ||
| ] |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Public boundary omits EventPublisher and ReviewPort.
All five runtime ports (CancellationPort, CheckpointPort, SecretResolver, EventPublisher, ReviewPort) are part of the application boundary per the PR objectives ("cancellation, checkpoints, secrets, events, and transform review"), but only three are re-exported here. New clients implementing a custom event publisher or reviewer would have to import from the internal loafer.ports.runtime module instead of this versioned boundary package.
♻️ Proposed fix
-from loafer.ports.runtime import CancellationPort, CheckpointPort, SecretResolver
+from loafer.ports.runtime import (
+ CancellationPort,
+ CheckpointPort,
+ EventPublisher,
+ ReviewPort,
+ SecretResolver,
+)
__all__ = [
"BatchEnvelope",
"CancellationPort",
"Checkpoint",
"CheckpointPort",
"ConnectorCatalog",
+ "EventPublisher",
"ExecutionPlan",
"LocalApplicationService",
"RunEvent",
"RunPipeline",
"RunRequest",
"RunResult",
"RunSnapshot",
"RunStatus",
+ "ReviewPort",
"SecretResolver",
"StageStatus",
"ValidationResult",
"get_local_application",
]📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| from loafer.ports.runtime import CancellationPort, CheckpointPort, SecretResolver | |
| __all__ = [ | |
| "BatchEnvelope", | |
| "CancellationPort", | |
| "Checkpoint", | |
| "CheckpointPort", | |
| "ConnectorCatalog", | |
| "ExecutionPlan", | |
| "LocalApplicationService", | |
| "RunEvent", | |
| "RunPipeline", | |
| "RunRequest", | |
| "RunResult", | |
| "RunSnapshot", | |
| "RunStatus", | |
| "SecretResolver", | |
| "StageStatus", | |
| "ValidationResult", | |
| "get_local_application", | |
| ] | |
| from loafer.ports.runtime import ( | |
| CancellationPort, | |
| CheckpointPort, | |
| EventPublisher, | |
| ReviewPort, | |
| SecretResolver, | |
| ) | |
| __all__ = [ | |
| "BatchEnvelope", | |
| "CancellationPort", | |
| "Checkpoint", | |
| "CheckpointPort", | |
| "ConnectorCatalog", | |
| "EventPublisher", | |
| "ExecutionPlan", | |
| "LocalApplicationService", | |
| "RunEvent", | |
| "RunPipeline", | |
| "RunRequest", | |
| "RunResult", | |
| "RunSnapshot", | |
| "RunStatus", | |
| "ReviewPort", | |
| "SecretResolver", | |
| "StageStatus", | |
| "ValidationResult", | |
| "get_local_application", | |
| ] |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@loafer/application/__init__.py` around lines 18 - 38, Update the public
exports in the application package’s __all__ to include the missing
EventPublisher and ReviewPort runtime ports, and import them from
loafer.ports.runtime alongside CancellationPort, CheckpointPort, and
SecretResolver. Preserve all existing boundary exports.
| def _stream_prepared( | ||
| self, | ||
| config: PipelineConfig, | ||
| plan: ExecutionPlan, | ||
| request: RunRequest, | ||
| ) -> Iterator[tuple[RunEvent, PipelineState]]: | ||
| updates = stream_pipeline( | ||
| config, | ||
| config_path=plan.config_path, | ||
| dry_run=request.dry_run, | ||
| auto_confirm=request.auto_confirm, | ||
| full_refresh=request.full_refresh, | ||
| run_id=request.run_id, | ||
| reviewer=self._reviewer, | ||
| secret_resolver=self._secrets, | ||
| provider_factory=self._provider_factory, | ||
| ) | ||
| sequence = 0 | ||
|
|
||
| while True: | ||
| if self._cancellation.is_cancelled(request.run_id): | ||
| raise PipelineError(f"Pipeline cancelled (run_id={request.run_id})") | ||
| try: | ||
| stage, status, state = next(updates) | ||
| except StopIteration: | ||
| return | ||
|
|
||
| sequence += 1 | ||
| event = RunEvent( | ||
| run_id=request.run_id, | ||
| plan_id=plan.plan_id, | ||
| sequence=sequence, | ||
| stage=stage, | ||
| status=StageStatus(status), | ||
| snapshot=_snapshot(state, plan, config), | ||
| ) | ||
| self._events.publish(event) | ||
| yield event, state |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Cancellation here triggers a cleanup-skip bug in stream_pipeline.
Raising PipelineError on self._cancellation.is_cancelled(...) abandons the updates generator returned by stream_pipeline (engine.py). See the comment on loafer/engine.py lines 298-317 for the root cause: cleanup (_cleanup_source_connector, _cleanup_elt_staging) is skipped when the generator is closed via GeneratorExit, which isn't caught by the except Exception clauses there. This affects every early-abandonment path through this loop, not just cancellation.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@loafer/application/service.py` around lines 246 - 283, Update stream_pipeline
so generator closure via GeneratorExit still executes _cleanup_source_connector
and _cleanup_elt_staging, rather than relying only on except Exception handlers.
Preserve _stream_prepared’s cancellation behavior and ensure all
early-abandonment paths finalize both cleanup operations before the generator
closes.
| try: | ||
| if dry_run: | ||
| yield from _stream_dry_run(graph, state, mode, start) | ||
| else: | ||
| yield from _stream_graph(graph, state, mode, start) | ||
| except PipelineError: | ||
| _cleanup_source_connector(state) | ||
| _cleanup_elt_staging(state, mode) | ||
| raise | ||
| except Exception as exc: | ||
| total_ms = (time.monotonic() - start) * 1000 | ||
| state["duration_ms"]["total"] = total_ms | ||
| _cleanup_source_connector(state) | ||
| _cleanup_elt_staging(state, mode) | ||
| raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc | ||
| else: | ||
| _cleanup_source_connector(state) | ||
| _cleanup_elt_staging(state, mode) | ||
| if not dry_run: | ||
| _persist_cursor(state) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
Does Python guarantee a generator's finally block runs when the generator is garbage collected via GeneratorExit?
💡 Result:
Python does not provide an absolute guarantee that a generator's finally block will execute upon garbage collection [1][2][3]. While Python attempts to clean up generators by calling their close method—which raises a GeneratorExit exception at the point of suspension—this process is subject to several limitations [4][1][5]. The execution of finally blocks in this context depends on the following factors: 1. Garbage Collection Timing: Garbage collection is not guaranteed to occur at any specific time [1]. If the generator object is never garbage collected, the finally block will not run [1][3]. This can happen if the generator remains reachable, such as being part of a reference cycle that the collector does not reclaim or simply because the program terminates before collection occurs [1][6]. 2. Generator State and Behavior: If a generator catches GeneratorExit and attempts to yield again, it violates the expected cleanup protocol, and Python will raise a RuntimeError [7][5][8]. This effectively prevents the generator from continuing to execute, which may leave the finally block uncompleted or aborted [1][3]. 3. Implementation Details: While CPython's reference counting typically triggers cleanup immediately when an object goes out of scope, this is an implementation detail and not a language-level guarantee across all Python implementations [6]. In summary, while Python makes a best-effort attempt to run finally blocks by injecting GeneratorExit when a generator is garbage collected, it cannot guarantee that the generator will ever be garbage collected or that it will complete the finally block successfully [1][2]. [4][1][2][3][6]
Citations:
- 1: https://stackoverflow.com/questions/71933885/when-is-finally-run-in-a-python-generator
- 2: https://peps.python.org/pep-0325/
- 3: https://stackoverflow.com/questions/49262379/does-finally-always-execute-in-python
- 4: https://peps.python.org/pep-0342/
- 5: https://stackoverflow.com/questions/30862196/generatorexit-in-python-generator
- 6: https://stackoverflow.com/questions/50091553/python-generators-garbage-collection
- 7: https://amir.rachum.com/generator-cleanup/
- 8: https://amirrachum.com/generator-cleanup/
🏁 Script executed:
set -e
printf '\n== engine.py outline ==\n'
ast-grep outline loafer/engine.py --view expanded
printf '\n== relevant slices ==\n'
sed -n '240,340p' loafer/engine.py | cat -n
printf '\n== cleanup helpers ==\n'
rg -n "def _cleanup_source_connector|def _cleanup_elt_staging|def _stream_graph|def _stream_dry_run|def stream_pipeline" loafer/engine.py loafer/application/service.py
printf '\n== service cancellation path ==\n'
sed -n '1,240p' loafer/application/service.py | cat -nRepository: lupppig/loafer
Length of output: 2011
🏁 Script executed:
python3 - <<'PY'
import pathlib, re
engine = pathlib.Path('loafer/engine.py').read_text()
service = pathlib.Path('loafer/application/service.py').read_text()
for name in ['_cleanup_source_connector', '_cleanup_elt_staging', '_stream_graph', '_stream_dry_run']:
m = re.search(rf'^def {name}\b.*?(?=^def |\Z)', engine, re.S | re.M)
print(f'## {name}')
if m:
text = m.group(0)
print('contains yield:', 'yield' in text)
print(text[:1200])
else:
print('not found')
print('## stream_pipeline is generator:', 'yield from _stream_dry_run' in engine and 'yield from _stream_graph' in engine)
print('## service has cancellation checks:', '_stream_prepared' in service and 'PipelineError' in service)
PYRepository: lupppig/loafer
Length of output: 2011
Move cleanup into finally Cleanup is skipped when this generator is closed early, so the source connector and ELT staging tables can be left behind. Keep _persist_cursor(...) on the success path, but move _cleanup_source_connector(...) and _cleanup_elt_staging(...) into a finally block.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@loafer/engine.py` around lines 298 - 317, Restructure the cleanup around the
streaming logic in the enclosing generator so _cleanup_source_connector(state)
and _cleanup_elt_staging(state, mode) execute from a finally block, including
when iteration closes early. Preserve the existing PipelineError and
generic-exception handling, and keep _persist_cursor(state) only on the
successful non-dry-run path.
| This is deliberately not a persistence contract: it may contain live | ||
| connectors, iterators, providers, and review callbacks. Durable clients use | ||
| the sanitized contracts in :mod:`loafer.application.contracts`. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Docstring references the wrong module path.
Says durable clients use contracts in :mod:loafer.application.contracts``, but the actual module is loafer.contracts (confirmed by every import site in this PR: `ports/runtime.py`, `application/init.py`, `application/service.py`).
📝 Proposed fix
This is deliberately not a persistence contract: it may contain live
connectors, iterators, providers, and review callbacks. Durable clients use
-the sanitized contracts in :mod:`loafer.application.contracts`.
+the sanitized contracts in :mod:`loafer.contracts`.📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| This is deliberately not a persistence contract: it may contain live | |
| connectors, iterators, providers, and review callbacks. Durable clients use | |
| the sanitized contracts in :mod:`loafer.application.contracts`. | |
| This is deliberately not a persistence contract: it may contain live | |
| connectors, iterators, providers, and review callbacks. Durable clients use | |
| the sanitized contracts in :mod:`loafer.contracts`. |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@loafer/graph/state.py` around lines 6 - 8, Correct the module reference in
the state module docstring to point to loafer.contracts instead of
loafer.application.contracts, leaving the surrounding persistence-contract
description unchanged.
1 participant
What changed
and results.
transform review.
Why
This establishes a stable boundary shared by local clients while keeping execution internals,
credentials, and ephemeral runtime state out of durable contracts. It prepares the codebase for
bounded batch execution and durable recovery without coupling those capabilities to the CLI or
scheduler.
Impact
Existing CLI, scheduler, and Python runner entry points remain compatible. New clients can use
LocalApplicationServiceand consume serializable plans, events, snapshots, and results.Validation
uv run pytest: 683 passed, 50 skippeduv run ruff check loafer testsuv run ruff format --check loafer testsSummary by CodeRabbit
New Features
Documentation
Tests