Engine 扫描调度链路 深度解析

821    def process_scan(self, scan: dict[str, Any], worker_id: int = 1) -> bool:
822        scan_id = int(scan["id"])
823        did_work = False
824        with self.db.connect() as conn:
825            workflow = self.db.load_workflow(conn, int(scan["workflow_id"]))
826
827        while True:
828            with self.db.connect() as conn:
829                current = self.db.load_scan(conn, scan_id)
830                if not current or current["status"] in NON_RUNNABLE_SCAN_STATUSES:
831                    return did_work
832                if not self._worker_can_pick_job(worker_id):
833                    return did_work
834                if current["status"] == PREWARMING_SCAN_STATUS:
835                    conn.commit()
836                    jobs = None
837                    completed = claimed = set()
838                elif current["status"] == "post_processing":
839                    conn.commit()
840                    jobs = None
841                    completed = claimed = set()
842                else:
843                    completed = self.db.load_completed_metadata(conn, scan_id)
844                    claimed = self.db.load_claimed_metadata(conn, scan_id)
845                    step_results = self.db.load_step_results(conn, scan_id)
846                    jobs = build_pending_jobs(
847                        scan=current,
848                        workflow=workflow,
849                        completed=completed,
850                        claimed=claimed,
851                        step_results=step_results,
852                    )
853                    conn.commit()
854
855            needs_new_container = current["status"] == "post_processing" or bool(jobs)
856            if needs_new_container and not self._new_scan_container_allowed(scan_id):
857                return did_work
858
859            harness = harness_for(
860                normalize_harness_name(current["harness"]),
861                timeout_seconds=self.runtime_harness_timeout_seconds(),
862                model_provider=scan_model_provider(current),
863                codex_model_provider=getattr(self.config, "codex_model_provider", None),
864                codex_cli_gate=self.codex_cli_gate,
865            )
866
867            if current["status"] == PREWARMING_SCAN_STATUS:
868                if self._ensure_scan_cache_prewarmed(current, restore_status="running"):
869                    did_work = True
870                    continue
871
872            if current["status"] != "post_processing" and jobs:
873                if self._ensure_scan_cache_prewarmed(current, restore_status="running"):
874                    did_work = True
875                    continue
876
877            if current["status"] == "post_processing":
878                if self._ensure_scan_cache_prewarmed(current, restore_status="post_processing"):
879                    did_work = True
880                    continue
881                if not self._worker_can_pick_job(worker_id):
882                    return did_work
883                did_post_work = self.post_processor.process_once(current, harness)
884                if did_post_work:
885                    return True
886                return did_work
887
888            if not jobs:
889                if claimed != completed:
890                    return did_work
891                with self.db.connect() as conn:
892                    transitioned = self.db.set_scan_status_if_current(conn, scan_id, "running", "post_processing")
893                    conn.commit()
894                if not transitioned:
895                    return did_work
896                LOGGER.info("scan %s workflow completed; starting post-processing", scan_id)
897                did_work = True
898                continue
899
900            did_claim = False
901            for job in jobs:
902                if not self._worker_can_pick_job(worker_id):
903                    return did_work
904                did_claim = self.execute_job(
905                    scan=current,
906                    workflow_id=workflow.id,
907                    job=job,
908                    harness=harness,
909                )
910                if did_claim:

91    for depth in workflow.depths:
92        steps = workflow.steps_at_depth(depth)
93        next_states: list[State] = []
94        depth_complete = previous_depth_complete
95
96        if _depth_consumes_all(steps, depth):
97            # Do not start a batch until every branch in the previous depth has
98            # completed. Otherwise a concurrent worker could run it over a
99            # partial result set.
100            input_states = [_batch_state(scan, states, depth - 1)] if previous_depth_complete and states else []
101        else:
102            input_states = states
103
104        for state in input_states:
105            for step in steps:
106                task_complete = True
107                task_next_states: list[State] = []
108                for repeat_run in range(1, runs + 1):
109                    repeated_state = _state_for_repeat(state, repeat_run)
110                    key = metadata_key(step.id, repeated_state)
111                    if key not in completed:
112                        task_complete = False
113                        depth_complete = False
114                        if key not in claimed:
115                            pending.append(Job(step=step, state=repeated_state))
116                        # The next repeat needs the output from this one, and the
117                        # next depth needs the complete accumulated task output.
118                        break
119                    task_next_states.extend(_next_states(step, repeated_state, step_results))
120                if task_complete:
121                    next_states.extend(task_next_states)
122
123        states = next_states
124        previous_depth_complete = depth_complete
125
126    return sorted(
127        pending,
128        key=lambda job: (-job.depth, job.state.repeat_run, job.step.order, job.state.prev_id),
129    )

915    def execute_job(self, *, scan, workflow_id, job, harness):
916        step = job.step
917        state = job.state
918        metadata_id = None
919        checked_out_commit = None
920        prompt_filled = ""
921        prepared = None
922        try:
923            schema = output_schema(step.output_format, step.multi_output)
924            thinking_effort = scan.get("thinking_effort") or "medium"
925            harness_name = normalize_harness_name(scan["harness"])
926            model_provider = scan_model_provider(scan)
927            with self.workspace_setup_slots:
928                started = now_utc()
929                with self.db.connect() as conn:
930                    current_scan = self.db.load_scan(conn, int(scan["id"]))
931                    if not current_scan or current_scan["status"] in NON_RUNNABLE_SCAN_STATUSES:
932                        conn.commit()
933                        return False
934                    metadata_id = self.db.claim_step_metadata(
935                        conn,
936                        scan_id=int(scan["id"]),
937                        workflow_id=workflow_id,
938                        step_id=step.id,
939                        prev_id=state.prev_id,
940                        prev_table=state.prev_table,
941                        repeat_run=state.repeat_run,
942                        prompt_template=step.content,
943                        prompt_filled="",
944                        checked_out_commit=None,
945                        run_started_at=started,
946                        model=scan["model"],
947                        harness=harness_name,
948                        thinking_effort=thinking_effort,
949                        model_provider=model_provider,
950                    )
951                    agent_skills = (
952                        self.db.load_agent_skills(conn, current_scan) if hasattr(self.db, "load_agent_skills") else []
953                    )
954                    load_prior_results = getattr(self.db, "load_prior_repeat_results", None)
955                    prior_repeat_results = (
956                        load_prior_results(
957                            conn,
958                            scan_id=int(scan["id"]),
959                            step_id=step.id,
960                            prev_id=state.prev_id,
961                            prev_table=state.prev_table,
962                            repeat_run=state.repeat_run,
963                        )
964                        if state.repeat_run > 1 and callable(load_prior_results)
965                        else []
966                    )
967                    conn.commit()
968
969                if metadata_id is None:
970                    return False
971
972                try:
973                    prepared = prepare_dependency_workspace(
974                        data_dir=self.config.data_dir,
975                        checkout_cache_dir=getattr(self.config, "checkout_cache_dir", None),
976                        metadata_id=metadata_id,
977                        scan=scan,
978                        github_token=self.config.github_token,
979                        agent_skills=agent_skills,
980                        harness_name=harness_name,
981                        model_provider=model_provider,
982                    )
983                    checked_out_commit = prepared.checked_out_commit
984                    context = {**state.context, **workspace_context(prepared)}
985                    rendered_prompt = render_prompt(step.content, context)
986                    prompt_parts = [
987                        native_agent_skills_prompt(agent_skills, harness_name),
988                        workspace_prompt_context(prepared.layout, prepared.manifest_json),
989                        rendered_prompt,
990                        repeat_append_prompt(state.repeat_run, prior_repeat_results),
991                    ]
992                    prompt_filled = harness_prompt(
993                        "\n\n".join(part for part in prompt_parts if part),
994                        multi_output=step.multi_output,
995                        schema=schema,
996                    )
997                    with self.db.connect() as conn: