Add orchestration adapters and orchestration-aware execution dispatch#86
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
…egrate-orchestration-system-options
|
| GitGuardian id | GitGuardian status | Secret | Commit | Filename | |
|---|---|---|---|---|---|
| 27568531 | Triggered | Username Password | b74dd4b | backend/tests/test_security.py | View secret |
🛠 Guidelines to remediate hardcoded secrets
- Understand the implications of revoking this secret by investigating where it is used in your code.
- Replace and store your secret safely. Learn here the best practices.
- Revoke and rotate this secret.
- If possible, rewrite git history. Rewriting git history is not a trivial act. You might completely break other contributing developers' workflow and you risk accidentally deleting legitimate data.
To avoid such incidents in the future consider
- following these best practices for managing and storing secrets including API keys and other credentials
- install secret detection on pre-commit to catch secret before it leaves your machine and ease remediation.
🦉 GitGuardian detects secrets in your source code to help developers and security teams secure the modern development process. You are seeing this because you or someone else with access to this repository has authorized GitGuardian to scan your pull request.
fuzziecoder
merged commit 5fe2a2d
into
codex/fix-remaining-issues-and-raise-pr
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| execution = initialize_execution(execution_data.pipeline_id, user_id=current_user.user_id) | ||
|
|
There was a problem hiding this comment.
🔴 Old code not removed in create_execution: second initialize_execution call and unconditional background task overwrite all orchestration work
After the new orchestration dispatch logic (lines 215-249), the original pre-PR code at lines 251-253 was never removed. This causes two critical problems:
Root Cause and Impact
Double execution creation: Line 219 creates the first execution with orchestration context, and the orchestration dispatch logic (lines 221-249) sets up logs, metadata, and status on it. But then line 251 calls initialize_execution again, creating a second Execution and reassigning the execution variable. The first execution becomes an orphan in executions_db — all orchestration metadata, logs, and status changes are lost.
# Line 219: First execution created with orchestration context
execution = initialize_execution(execution_data.pipeline_id, context=orchestration_context)
# Lines 221-249: Orchestration dispatch sets metadata on `execution`...
# Line 251: OVERWRITES `execution` with a brand-new one — all work above is discarded
execution = initialize_execution(execution_data.pipeline_id, user_id=current_user.user_id)Unconditional local background task: Line 253 always calls background_tasks.add_task(execute_pipeline_background, ...), even when the user selected an external engine like Temporal or Prefect. This means every execution runs locally regardless of the chosen orchestration engine, completely defeating the purpose of the new feature.
# Line 253: Always runs, even for temporal/prefect/airflow
background_tasks.add_task(execute_pipeline_background, execution_data.pipeline_id, execution.id)Impact: The orchestration feature is entirely broken. External engines will never be used in practice, orchestration metadata is always lost, and orphan execution records accumulate in executions_db.
(Refers to lines 251-253)
Prompt for agents
In backend/api/routes/executions.py, the create_execution function (starting at line 203) has leftover pre-PR code at lines 251-253 that must be removed. These three lines are:
execution = initialize_execution(execution_data.pipeline_id, user_id=current_user.user_id)
background_tasks.add_task(execute_pipeline_background, execution_data.pipeline_id, execution.id)
return execution
They should be deleted entirely. The single return statement should just be `return execution` after the if/else orchestration block (after line 249).
Additionally, the first initialize_execution call at line 219 is missing the required user_id argument. It should be changed from:
execution = initialize_execution(execution_data.pipeline_id, context=orchestration_context)
to:
execution = initialize_execution(execution_data.pipeline_id, user_id=current_user.user_id, context=orchestration_context)
The corrected create_execution body (after the pipeline ownership check) should be:
orchestration_context = _build_orchestration_context(execution_data)
execution = initialize_execution(execution_data.pipeline_id, user_id=current_user.user_id, context=orchestration_context)
orchestration_engine = ORCHESTRATION_SCHEMA_TO_CORE[execution_data.orchestration.engine]
if orchestration_engine == OrchestrationEngine.LOCAL:
background_tasks.add_task(
execute_pipeline_background,
execution_data.pipeline_id,
execution.id
)
else:
pipeline = pipelines_db[execution_data.pipeline_id]
orchestration_request = OrchestrationRequest(
engine=orchestration_engine,
execution_id=execution.id,
pipeline_id=execution.pipeline_id,
pipeline_name=pipeline.name,
retry_attempts=execution_data.orchestration.retry_attempts,
retry_backoff_seconds=execution_data.orchestration.retry_backoff_seconds,
schedule=execution_data.orchestration.schedule,
options=execution_data.orchestration.options.copy(),
)
orchestration_result = OrchestrationRegistry().get(orchestration_engine).dispatch(orchestration_request)
execution.add_log(
None,
"INFO",
orchestration_result.message,
metadata=orchestration_result.metadata,
)
execution.context.setdefault("orchestration", {}).update(orchestration_result.metadata)
execution.status = orchestration_result.status
return execution
Also fix the broken import block at lines 10-27: the new imports (OrchestrationEngineSchema, SuccessResponse, and the orchestration module imports) must replace rather than sit alongside the old import lines. The dangling lines 20-22 (ExecutionResponse, SuccessResponse, closing paren) and duplicate imports on lines 24-27 need to be consolidated into a single clean import block.
Was this helpful? React with 👍 or 👎 to provide feedback.
| orchestration_context = _build_orchestration_context(execution_data) | ||
|
|
||
| # Create execution record | ||
| execution = initialize_execution(execution_data.pipeline_id, context=orchestration_context) |
There was a problem hiding this comment.
🔴 First initialize_execution call missing required user_id — execution created without owner
At line 219, the new code calls initialize_execution(execution_data.pipeline_id, context=orchestration_context) but omits the required user_id argument.
Detailed Explanation
The initialize_execution function signature at backend/api/routes/executions.py:89-92 is:
def initialize_execution(
pipeline_id: str,
user_id: str,
context: Optional[Dict[str, Any]] = None,
) -> Execution:user_id is a required positional parameter. The call at line 219:
execution = initialize_execution(execution_data.pipeline_id, context=orchestration_context)is missing user_id. Even if this weren't caught at runtime (it would be a TypeError), the Execution dataclass at backend/models/pipeline.py:86-98 requires user_id: str as a non-optional field. The intent is clearly to pass current_user.user_id so that the execution is properly scoped to the authenticated user for authorization checks on subsequent reads.
Impact: Without user_id, the execution would not be associated with any user, breaking all user-scoped queries (list, get, cancel) which filter by e.user_id == current_user.user_id.
| execution = initialize_execution(execution_data.pipeline_id, context=orchestration_context) | |
| execution = initialize_execution(execution_data.pipeline_id, user_id=current_user.user_id, context=orchestration_context) |
Was this helpful? React with 👍 or 👎 to provide feedback.
| pipeline = Pipeline( | ||
| id="pipe-orchestration", | ||
| name="Orchestration Pipeline", | ||
| description="Pipeline for orchestration tests", | ||
| stages=[ | ||
| Stage(id="input", name="Input", type=StageType.INPUT, config={"source": "test", "data": [1]}, dependencies=[]), | ||
| Stage(id="output", name="Output", type=StageType.OUTPUT, config={"destination": "sink"}, dependencies=["input"]), | ||
| ], | ||
| ) |
There was a problem hiding this comment.
🔴 Test _seed_pipeline missing required user_id field on Pipeline dataclass
The test helper _seed_pipeline() creates a Pipeline without providing the required user_id field.
Detailed Explanation
At backend/tests/test_orchestration_engine.py:19-27, the pipeline is created as:
pipeline = Pipeline(
id="pipe-orchestration",
name="Orchestration Pipeline",
description="Pipeline for orchestration tests",
stages=[...],
)But the Pipeline dataclass at backend/models/pipeline.py:50-59 requires user_id: str as a non-optional positional field (it comes before the fields with defaults). This will cause a TypeError when the test tries to instantiate the Pipeline, making all three new tests fail.
Additionally, the create_execution endpoint at backend/api/routes/executions.py:210 checks pipeline.user_id != current_user.user_id, so even if Pipeline allowed a missing user_id, the ownership check would fail.
| pipeline = Pipeline( | |
| id="pipe-orchestration", | |
| name="Orchestration Pipeline", | |
| description="Pipeline for orchestration tests", | |
| stages=[ | |
| Stage(id="input", name="Input", type=StageType.INPUT, config={"source": "test", "data": [1]}, dependencies=[]), | |
| Stage(id="output", name="Output", type=StageType.OUTPUT, config={"destination": "sink"}, dependencies=["input"]), | |
| ], | |
| ) | |
| pipeline = Pipeline( | |
| id="pipe-orchestration", | |
| name="Orchestration Pipeline", | |
| description="Pipeline for orchestration tests", | |
| user_id="test-user", | |
| stages=[ | |
| Stage(id="input", name="Input", type=StageType.INPUT, config={"source": "test", "data": [1]}, dependencies=[]), | |
| Stage(id="output", name="Output", type=StageType.OUTPUT, config={"destination": "sink"}, dependencies=["input"]), | |
| ], | |
| ) |
Was this helpful? React with 👍 or 👎 to provide feedback.
1 participant
Motivation
localexecution as the default to preserve backward compatibility.Description
backend/core/orchestration.pyimplementingOrchestrationEngine,OrchestrationRequest,OrchestrationResult, adapters forAirflowOrchestrator,TemporalOrchestrator,PrefectOrchestrator, and anOrchestrationRegistry.backend.coreviabackend/core/__init__.pyfor easier imports.ExecutionCreateinbackend/api/schemas.pywith anOrchestrationConfig(fields:engine,retry_attempts,retry_backoff_seconds,schedule,options) and addedOrchestrationEngineSchemaenum.backend/api/routes/executions.pyto build orchestration context, persist it on theExecution, dispatch to external orchestrators when selected, or run the existing localexecute_pipeline_backgroundforlocal, and to record orchestration metadata in logs/context.backend/tests/test_orchestration_engine.pyto validatelocal,temporal, andprefectdispatch behaviors and updated behavior integration with existing Airflow tests.Testing
PYTHONPATH=. pytest -q backend/tests/test_orchestration_engine.py backend/tests/test_airflow_integration.pyand they passed (8 passed, multiple warnings about deprecations only).contextand thatlocalstill enqueues a background task while external engines setpendingand include provider-specific metadata.Codex Task