fix: harden generated execution authority

src/blackcell/adapters/execution/worktree.py

@@ -563,6 +563,40 @@ def validate_base_commit(self, repository_root: Path, base_commit: str) -> None:
         if commit.return_code != 0:
             raise WorktreeLifecycleError(WorktreeFailureCode.BASE_COMMIT_NOT_FOUND)
 
+    def changed_paths_between(
+        self,
+        repository_root: Path,
+        *,
+        base_commit: str,
+        head_commit: str,
+    ) -> tuple[str, ...]:
+        """Return normalized cumulative path effects for one admitted commit range."""
+
+        root = _canonical_repository_root(repository_root)
+        self.validate_base_commit(root, base_commit)
+        self.validate_base_commit(root, head_commit)
+        ancestor = self._git_at(
+            root,
+            ("merge-base", "--is-ancestor", base_commit, head_commit),
+        )
+        if ancestor.return_code != 0:
+            raise WorktreeLifecycleError(WorktreeFailureCode.WORKTREE_CONFLICT)
+        changed = self._require_success(
+            self._git_at(
+                root,
+                (
+                    "diff",
+                    "--name-only",
+                    "--no-renames",
+                    "-z",
+                    base_commit,
+                    head_commit,
+                    "--",
+                ),
+            )
+        )
+        return self._path_output(changed)
+
     def retain_plan_base_commit(
         self,
         repository_root: Path,

src/blackcell/bootstrap/execution_process.py

@@ -62,7 +62,9 @@
     MAX_CHANGE_PROPOSAL_BYTES,
 )
 from blackcell.orchestration.execution_plan import (
+    ExecutionAuthority,
     ExecutionPolicyKernel,
+    GoalSpec,
     PlanningRequest,
     RunLifecycleStatus,
     planning_payload,
@@ -101,6 +103,10 @@ def run_once(self) -> ExecutionWorkerCycleResult: ...
 class ExecutionReconciliationPort(Protocol):
     def next_generated_run(self) -> GeneratedRun | None: ...
 
+    def generated_execution_goal(self, run_id: str) -> GoalSpec: ...
+
+    def generated_execution_authority(self, run_id: str) -> ExecutionAuthority: ...
+
     def should_cancel_generated_run(self, run_id: str) -> bool: ...
 
     def reconcile_startup(self, *, principal_id: str) -> tuple[object, ...]: ...
@@ -202,6 +208,7 @@ def from_config(
                 evidence=ExecutionEvidenceCollector(boundaries.worktrees),
                 changes=TextChangeExecutor(boundaries.worktrees),
                 cancel_requested=runtime.should_cancel_generated_run,
+                authority_for_run=runtime.generated_execution_authority,
             )
             observer = (
                 None if telemetry.recorder is None else ExecutionTraceObserver(telemetry.recorder)
@@ -216,7 +223,9 @@ def from_config(
                     boundaries.planner,
                     execution_executor,
                     ExecutionPolicyKernel(),
-                )
+                ),
+                authority_for_run=runtime.generated_execution_authority,
+                goal_for_run=runtime.generated_execution_goal,
             )
         except Exception:
             telemetry.shutdown()
@@ -253,11 +262,14 @@ def _run_generated_once(
         generated = self.runtime.next_generated_run()
         if generated is None:
             return None
-        budget = GatewayBudget(
-            config.provider.max_input_tokens,
-            config.provider.max_output_tokens,
-            config.provider.timeout_ceiling_seconds * 1_000,
-            config.provider.max_cost_microusd,
+        budget = _intersect_budget(
+            GatewayBudget(
+                config.provider.max_input_tokens,
+                config.provider.max_output_tokens,
+                config.provider.timeout_ceiling_seconds * 1_000,
+                config.provider.max_cost_microusd,
+            ),
+            generated.authority.budget,
         )
         payload_size = len(canonical_json_bytes(planning_payload(generated.goal)))
         request = PlanningRequest(
@@ -336,6 +348,15 @@ def serve(self, *, once: bool = False) -> int:
             self.shutdown()
 
 
+def _intersect_budget(left: GatewayBudget, right: GatewayBudget) -> GatewayBudget:
+    return GatewayBudget(
+        min(left.max_input_tokens, right.max_input_tokens),
+        min(left.max_output_tokens, right.max_output_tokens),
+        min(left.max_latency_ms, right.max_latency_ms),
+        min(left.max_cost_microusd, right.max_cost_microusd),
+    )
+
+
 def validate_execution_worker_runtime_config(
     config: RuntimeProcessConfig,
     *,