Skip to content

feat(core): ExecutionContext abstraction for task isolation (#532) #536

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
frankbria merged 2 commits into from
Apr 3, 2026
Merged

feat(core): ExecutionContext abstraction for task isolation (#532) #536

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
01acead
feat(core): ExecutionContext abstraction for task isolation (#532)
Apr 3, 2026
e1a95c6
fix(tests): use git init -b main to set explicit branch name in CI
Apr 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 16 additions & 1 deletion codeframe/cli/app.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2319,6 +2319,12 @@ def work_start(
help="Recovery action on stall: 'blocker' (default), 'retry', or 'fail'",
click_type=click.Choice(["blocker", "retry", "fail"], case_sensitive=False),
),
isolation: str = typer.Option(
"none",
"--isolation",
help="Task execution isolation: none (default), worktree, or cloud",
click_type=click.Choice(["none", "worktree", "cloud"], case_sensitive=False),
),
) -> None:
"""Start working on a task.

Expand All @@ -2331,6 +2337,7 @@ def work_start(
codeframe work start abc123 --execute --engine plan
codeframe work start abc123 --execute --dry-run
codeframe work start abc123 --execute --verbose
codeframe work start abc123 --execute --isolation worktree
"""
from codeframe.core.workspace import get_workspace
from codeframe.core import tasks as tasks_module, runtime
Expand Down Expand Up @@ -2398,7 +2405,7 @@ def work_start(
state = runtime.execute_agent(
workspace, run, dry_run=dry_run, debug=debug, verbose=verbose,
engine=engine, stall_timeout_s=stall_timeout,
stall_action=stall_action,
stall_action=stall_action, isolation=isolation,
)

if state.status == AgentStatus.COMPLETED:
Expand Down Expand Up @@ -3550,6 +3557,12 @@ def batch_run(
help="Recovery action on stall: 'blocker' (default), 'retry', or 'fail'",
click_type=click.Choice(["blocker", "retry", "fail"], case_sensitive=False),
),
isolation: str = typer.Option(
"none",
"--isolation",
help="Task execution isolation: none (default), worktree, or cloud",
click_type=click.Choice(["none", "worktree", "cloud"], case_sensitive=False),
),
) -> None:
"""Execute multiple tasks in batch.

Expand All @@ -3565,6 +3578,7 @@ def batch_run(
codeframe work batch run --all-ready --engine plan
codeframe work batch run task1 task2 --dry-run
codeframe work batch run task1 task2 --retry 2
codeframe work batch run --all-ready --isolation worktree
"""
from codeframe.core.workspace import get_workspace
from codeframe.core import tasks as tasks_module, conductor
Expand Down Expand Up @@ -3663,6 +3677,7 @@ def batch_run(
engine=engine,
stall_timeout_s=stall_timeout,
stall_action=stall_action,
isolation=isolation,
)

# Show summary
Expand Down
133 changes: 107 additions & 26 deletions codeframe/core/conductor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -538,6 +538,7 @@ class BatchRun:
stall_action: str = "blocker"
concurrency: ConcurrencyConfig = field(default_factory=ConcurrencyConfig)
isolate: bool = True
isolation: str = "none"


def start_batch(
Expand All @@ -554,6 +555,7 @@ def start_batch(
stall_action: str = "blocker",
concurrency_by_status: Optional[dict[str, int]] = None,
isolate: bool = True,
isolation: str = "none",
) -> BatchRun:
Comment on lines 557 to 559

@coderabbitai coderabbitai Bot Apr 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Validate isolation before the batch is persisted.

isolation is stored as a raw string here and only converted inside the execution loops. If callers pass "cloud" or any invalid value, the first create_execution_context(...) raises after the batch has already been saved and moved to RUNNING, which leaves a half-started batch instead of a clean validation error.

🛠️ Suggested preflight validation 
 def start_batch(
@@
     if not task_ids:
         raise ValueError("task_ids cannot be empty")
+
+    from codeframe.core.sandbox.context import IsolationLevel
+
+    try:
+        isolation_level = IsolationLevel(isolation)
+    except ValueError as exc:
+        raise ValueError(f"Unsupported isolation level: {isolation}") from exc
+    if isolation_level is IsolationLevel.CLOUD:
+        raise ValueError(
+            "Isolation level 'cloud' is not implemented yet. Use 'none' or 'worktree'."
+        )
@@
         concurrency=concurrency,
         isolate=isolate,
-        isolation=isolation,
+        isolation=isolation_level.value,
     )

Also applies to: 598-615

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@codeframe/core/conductor.py` around lines 557 - 559, Validate the isolation
parameter before persisting the batch: in the function that takes isolate: bool
and isolation: str -> BatchRun, check that isolation is one of the allowed
values (e.g., "none", "host", "container", "vm" or your project's canonical set)
and raise a clear validation error if not, before calling
create_execution_context(...) or saving/moving the batch to RUNNING; apply the
same preflight check for the similar block around create_execution_context usage
at the other location referenced (the code between the earlier
create_execution_context call and the block at 598-615) so no batch is persisted
when isolation is invalid.

"""Start a batch execution of multiple tasks.

Expand Down Expand Up @@ -609,6 +611,7 @@ def start_batch(
stall_action=stall_action,
concurrency=concurrency,
isolate=isolate,
isolation=isolation,
)

# Save to database
Expand Down Expand Up @@ -1014,8 +1017,17 @@ def _execute_serial_resume(
if on_event:
on_event("batch_task_started", {"task_id": task_id, "position": i + 1, "is_retry": True})

# Execute task via subprocess
result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action)
# Execute task via subprocess (with isolation context)
from codeframe.core.sandbox.context import IsolationLevel, create_execution_context
exec_ctx = create_execution_context(task_id, IsolationLevel(batch.isolation), workspace.repo_path)
try:
result_status = _execute_task_subprocess(
workspace, task_id, batch.id, engine=batch.engine,
stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action,
worktree_path=exec_ctx.workspace_path if exec_ctx.workspace_path != workspace.repo_path else None,
)
finally:
exec_ctx.cleanup()
Comment on lines +1020 to +1030

@coderabbitai coderabbitai Bot Apr 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

worktree runs currently delete their own outputs.

These paths tear the ExecutionContext down in finally, but there is no merge/copy/patch step before cleanup. Because task tools write against exec_ctx.workspace_path, a successful WORKTREE run leaves its edits in the task worktree and then deletes them here, so the batch can report success while workspace.repo_path stays unchanged. Harvest the worktree changes before cleanup, or keep the worktree around until a caller has done so.

Based on learnings, "agent tool functions ... must be stateless with signature (input_data: dict, workspace_path: Path, tool_call_id: str) -> ToolResult".

Also applies to: 1197-1207, 1426-1448, 1840-1870, 1960-1976

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@codeframe/core/conductor.py` around lines 1020 - 1030, The worktree created
by create_execution_context (ExecutionContext / exec_ctx) is torn down in the
finally block via exec_ctx.cleanup() without harvesting changes, so any edits
under exec_ctx.workspace_path are deleted; before calling exec_ctx.cleanup()
(after _execute_task_subprocess returns success or result_status is available)
detect when exec_ctx.workspace_path != workspace.repo_path and persist the
worktree edits back into the main workspace (e.g., by merging/patching/copying
changed files or invoking a commit/merge method on ExecutionContext); add or
call a clearly named method such as exec_ctx.harvest_changes() or
exec_ctx.merge_back_to(workspace.repo_path) prior to cleanup, and ensure
_execute_task_subprocess and any agent tool functions remain stateless and write
only to the provided workspace_path; apply the same harvest-before-cleanup fix
at the other similar locations referenced (around the other ExecutionContext
usages).


# Record result (overwrites previous result)
batch.results[task_id] = result_status
Expand Down Expand Up @@ -1182,8 +1194,17 @@ def _execute_retries(
print(f"\n[Retry {retry_num}, {i + 1}/{len(failed_tasks)}] {task_id}: {task_title}")
print(f" Previous: {previous_status}")

# Execute task
result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action)
# Execute task (with isolation context)
from codeframe.core.sandbox.context import IsolationLevel, create_execution_context
exec_ctx = create_execution_context(task_id, IsolationLevel(batch.isolation), workspace.repo_path)
try:
result_status = _execute_task_subprocess(
workspace, task_id, batch.id, engine=batch.engine,
stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action,
worktree_path=exec_ctx.workspace_path if exec_ctx.workspace_path != workspace.repo_path else None,
)
finally:
exec_ctx.cleanup()

# Update result
batch.results[task_id] = result_status
Expand Down Expand Up @@ -1402,16 +1423,29 @@ def _execute_serial(
if on_event:
on_event("batch_task_started", {"task_id": task_id, "position": i + 1})

# Execute task via subprocess
result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action)
# Execute task via subprocess (with isolation context)
from codeframe.core.sandbox.context import IsolationLevel, create_execution_context
exec_ctx = create_execution_context(task_id, IsolationLevel(batch.isolation), workspace.repo_path)
try:
result_status = _execute_task_subprocess(
workspace, task_id, batch.id, engine=batch.engine,
stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action,
worktree_path=exec_ctx.workspace_path if exec_ctx.workspace_path != workspace.repo_path else None,
)

# If task is BLOCKED, try supervisor resolution
if result_status == RunStatus.BLOCKED.value:
supervisor = get_supervisor(workspace)
if supervisor.try_resolve_blocked_task(task_id):
# Supervisor resolved the blocker - retry the task
print(" [Supervisor] Retrying task after auto-resolution...")
result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action)
# If task is BLOCKED, try supervisor resolution
if result_status == RunStatus.BLOCKED.value:
supervisor = get_supervisor(workspace)
if supervisor.try_resolve_blocked_task(task_id):
# Supervisor resolved the blocker - retry the task
print(" [Supervisor] Retrying task after auto-resolution...")
result_status = _execute_task_subprocess(
workspace, task_id, batch.id, engine=batch.engine,
stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action,
worktree_path=exec_ctx.workspace_path if exec_ctx.workspace_path != workspace.repo_path else None,
)
finally:
exec_ctx.cleanup()

# Record result
batch.results[task_id] = result_status
Expand Down Expand Up @@ -1803,16 +1837,37 @@ def _execute_single_task(
if on_event:
on_event("batch_task_started", {"task_id": task_id, "position": position})

# Execute task via subprocess
result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action)
# Create execution context (handles isolation level; NONE is a no-op)
from codeframe.core.sandbox.context import IsolationLevel, create_execution_context
exec_ctx = create_execution_context(
task_id, IsolationLevel(batch.isolation), workspace.repo_path
)

try:
# Execute task via subprocess
result_status = _execute_task_subprocess(
workspace, task_id, batch.id,
engine=batch.engine,
stall_timeout_s=batch.stall_timeout_s,
stall_action=batch.stall_action,
worktree_path=exec_ctx.workspace_path if exec_ctx.workspace_path != workspace.repo_path else None,
)

# If task is BLOCKED, try supervisor resolution
if result_status == RunStatus.BLOCKED.value:
supervisor = get_supervisor(workspace)
if supervisor.try_resolve_blocked_task(task_id):
# Supervisor resolved the blocker - retry the task
print(" [Supervisor] Retrying task after auto-resolution...")
result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action)
# If task is BLOCKED, try supervisor resolution
if result_status == RunStatus.BLOCKED.value:
supervisor = get_supervisor(workspace)
if supervisor.try_resolve_blocked_task(task_id):
# Supervisor resolved the blocker - retry the task
print(" [Supervisor] Retrying task after auto-resolution...")
result_status = _execute_task_subprocess(
workspace, task_id, batch.id,
engine=batch.engine,
stall_timeout_s=batch.stall_timeout_s,
stall_action=batch.stall_action,
worktree_path=exec_ctx.workspace_path if exec_ctx.workspace_path != workspace.repo_path else None,
)
finally:
exec_ctx.cleanup()

# Record result
batch.results[task_id] = result_status
Expand Down Expand Up @@ -1902,8 +1957,23 @@ def execute_task(task_id: str) -> tuple[str, str]:
if on_event:
on_event("batch_task_started", {"task_id": task_id, "parallel": True})

# Execute via subprocess
result_status = _execute_task_subprocess(workspace, task_id, batch.id, engine=batch.engine, stall_timeout_s=batch.stall_timeout_s, stall_action=batch.stall_action)
# Create execution context for this task
from codeframe.core.sandbox.context import IsolationLevel, create_execution_context
exec_ctx = create_execution_context(
task_id, IsolationLevel(batch.isolation), workspace.repo_path
)

try:
# Execute via subprocess
result_status = _execute_task_subprocess(
workspace, task_id, batch.id,
engine=batch.engine,
stall_timeout_s=batch.stall_timeout_s,
stall_action=batch.stall_action,
worktree_path=exec_ctx.workspace_path if exec_ctx.workspace_path != workspace.repo_path else None,
)
finally:
exec_ctx.cleanup()

# Record result (thread-safe due to GIL for simple dict operations)
batch.results[task_id] = result_status
Expand Down Expand Up @@ -2062,12 +2132,21 @@ def _save_batch(workspace: Workspace, batch: BatchRun) -> None:
conn = get_db_connection(workspace)
try:
cursor = conn.cursor()
# Ensure isolation column exists (migration for existing databases)
try:
cursor.execute(
"ALTER TABLE batch_runs ADD COLUMN isolation TEXT DEFAULT 'none'"
)
conn.commit()
except Exception:
pass # Column already exists

cursor.execute(
"""
INSERT OR REPLACE INTO batch_runs
(id, workspace_id, task_ids, status, strategy, max_parallel, on_failure,
started_at, completed_at, results, engine)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
started_at, completed_at, results, engine, isolation)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
Comment on lines 2146 to +2149

@coderabbitai coderabbitai Bot Apr 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Loaded batches always fall back to isolation="none".

_row_to_batch() now expects row[11], but get_batch() and list_batches() still select only ... results, engine from batch_runs. Every reloaded batch therefore drops its stored isolation, so resume_batch() re-executes on the shared repo even if the original batch used worktree.

💾 Suggested query fix 
         SELECT id, workspace_id, task_ids, status, strategy, max_parallel,
-               on_failure, started_at, completed_at, results, engine
+               on_failure, started_at, completed_at, results, engine, isolation
         FROM batch_runs
@@
             SELECT id, workspace_id, task_ids, status, strategy, max_parallel,
-                   on_failure, started_at, completed_at, results, engine
+                   on_failure, started_at, completed_at, results, engine, isolation
             FROM batch_runs
@@
             SELECT id, workspace_id, task_ids, status, strategy, max_parallel,
-                   on_failure, started_at, completed_at, results, engine
+                   on_failure, started_at, completed_at, results, engine, isolation
             FROM batch_runs

Also applies to: 2162-2163, 2171-2185

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@codeframe/core/conductor.py` around lines 2146 - 2149, The SELECTs that
populate rows for _row_to_batch() are missing the isolation column so loaded
batches default to "none"; update get_batch() and list_batches() (and any other
queries that fetch batch_runs rows used by _row_to_batch(), e.g., resume_batch
related queries) to include the isolation column in the SELECT list in the same
order as the INSERT (id, workspace_id, task_ids, status, strategy, max_parallel,
on_failure, started_at, completed_at, results, engine, isolation) so
_row_to_batch() reads isolation from row[11] correctly.

""",
(
batch.id,
Expand All @@ -2081,6 +2160,7 @@ def _save_batch(workspace: Workspace, batch: BatchRun) -> None:
completed_at,
results_json,
batch.engine,
batch.isolation,
),
)
conn.commit()
Expand All @@ -2102,4 +2182,5 @@ def _row_to_batch(row: tuple) -> BatchRun:
completed_at=datetime.fromisoformat(row[8]) if row[8] else None,
results=json.loads(row[9]) if row[9] else {},
engine=row[10] if len(row) > 10 and row[10] else "plan",
isolation=row[11] if len(row) > 11 and row[11] else "none",
)
14 changes: 12 additions & 2 deletions codeframe/core/runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -599,6 +599,7 @@ def execute_agent(
engine: str = "react",
stall_timeout_s: int = 300,
stall_action: str = "blocker",
isolation: str = "none",
) -> "AgentState":
Comment on lines +602 to 603

@coderabbitai coderabbitai Bot Apr 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Move isolation setup under the guarded execution path.

create_execution_context() runs before the main try, so --isolation cloud (currently NotImplementedError) or any worktree setup failure exits execute_agent() before fail_run() runs. That leaves the run/task stuck active even though execution never started. Initialize exec_ctx first, create the context inside the try, and only clean it up when setup succeeded.

🛠️ Suggested fix 
-    # Create execution context (handles isolation; NONE is a no-op)
-    from codeframe.core.sandbox.context import IsolationLevel, create_execution_context
-    exec_ctx = create_execution_context(
-        run.task_id, IsolationLevel(isolation), workspace.repo_path
-    )
-    effective_repo_path = exec_ctx.workspace_path
-
-    try:
+    # Create execution context (handles isolation; NONE is a no-op)
+    from codeframe.core.sandbox.context import IsolationLevel, create_execution_context
+    exec_ctx = None
+    effective_repo_path = workspace.repo_path
+
+    try:
+        exec_ctx = create_execution_context(
+            run.task_id, IsolationLevel(isolation), workspace.repo_path
+        )
+        effective_repo_path = exec_ctx.workspace_path
-        exec_ctx.cleanup()
+        if exec_ctx is not None:
+            exec_ctx.cleanup()

Also applies to: 679-685, 877-878

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@codeframe/core/runtime.py` around lines 602 - 603, The isolation/worktree
setup (create_execution_context) must be moved into the guarded execution path
so failures don't bypass fail_run(); change execute_agent so exec_ctx is
declared/initialized to None before the try, then call
create_execution_context(...) inside the try block (e.g., where execution
begins) and only perform cleanup if exec_ctx is not None; apply same pattern for
the other occurrences that initialize execution contexts (the blocks around
create_execution_context and cleanup mentioned near lines 679-685 and 877-878),
and ensure fail_run() still runs on exceptions.

"""Execute a task using the agent orchestrator.

Expand Down Expand Up @@ -675,6 +676,13 @@ def execute_agent(
import time as _time_mod
_perf_start_ms = int(_time_mod.monotonic() * 1000)

# Create execution context (handles isolation; NONE is a no-op)
from codeframe.core.sandbox.context import IsolationLevel, create_execution_context
exec_ctx = create_execution_context(
run.task_id, IsolationLevel(isolation), workspace.repo_path
)
effective_repo_path = exec_ctx.workspace_path

Comment on lines +679 to +685

@coderabbitai coderabbitai Bot Apr 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Create the context before building hook state.

Because the execution context is created after HookContext is assembled, Line 673 and the execute_hook(..., workspace.repo_path, ...) calls still point at the shared repo. With worktree isolation, before_task / after_task_* hooks can read or mutate the main checkout while the agent runs against the worktree, which breaks task isolation.

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@codeframe/core/runtime.py` around lines 679 - 685, The HookContext and hook
invocations are being built with the shared repo path before the execution
context is created, which breaks isolation; move the
create_execution_context(...) call (using IsolationLevel(isolation) and
workspace.repo_path) so exec_ctx = create_execution_context(...) runs before
constructing HookContext and before any execute_hook(...) calls, then use
exec_ctx.workspace_path (or effective_repo_path) as the repo path when building
HookContext and when calling execute_hook(...) so hooks operate against the
execution context workspace rather than the main checkout; keep the same
create_execution_context, IsolationLevel, HookContext, execute_hook, and
exec_ctx symbols when editing.

try:
# Execute before_task hook (aborts on failure)
if env_config and hook_ctx:
Expand Down Expand Up @@ -733,7 +741,7 @@ def on_adapter_event(event: AdapterEvent) -> None:
)

result = wrapper.run(
run.task_id, packaged.prompt, workspace.repo_path,
run.task_id, packaged.prompt, effective_repo_path,
on_event=on_adapter_event,
)
else:
Expand All @@ -758,7 +766,7 @@ def on_adapter_event(event: AdapterEvent) -> None:
)

result = adapter.run(
run.task_id, "", workspace.repo_path,
run.task_id, "", effective_repo_path,
on_event=on_adapter_event,
)

Expand Down Expand Up @@ -866,6 +874,8 @@ def on_adapter_event(event: AdapterEvent) -> None:
finally:
# Always close the output logger to ensure file is properly flushed
output_logger.close()
# Clean up execution context (no-op for NONE, removes worktree for WORKTREE)
exec_ctx.cleanup()


def _event_type_to_category(event_type: str):
Expand Down
13 changes: 13 additions & 0 deletions codeframe/core/sandbox/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
"""Execution environment sandbox abstraction.

Provides ExecutionContext and IsolationLevel for isolating task execution
from the shared filesystem.
"""

from codeframe.core.sandbox.context import (
ExecutionContext,
IsolationLevel,
create_execution_context,
)

__all__ = ["ExecutionContext", "IsolationLevel", "create_execution_context"]
Loading
Loading