From f8c690999a6f40964bac9075a34f9c0fba40f7a3 Mon Sep 17 00:00:00 2001
From: Marius Ciepluch <11855163+norandom@users.noreply.github.com>
Date: Thu, 18 Jun 2026 07:57:06 +0200
Subject: [PATCH] feat(harness): add runtime and reachability mapping

---
 .kiro/specs/openrouter-sast-harness/design.md |  36 +-
 .../openrouter-sast-harness/requirements.md   |  50 +-
 .kiro/specs/openrouter-sast-harness/tasks.md  | 115 ++++-
 src/openultrasast/cli.py                      |  38 +-
 src/openultrasast/config.py                   |  15 +
 src/openultrasast/findings.py                 |  65 ++-
 src/openultrasast/harness.py                  | 187 +++++++
 src/openultrasast/mapping.py                  | 455 ++++++++++++++++++
 src/openultrasast/preprocess.py               |  27 +-
 src/openultrasast/rank.py                     |  23 +
 src/openultrasast/reports.py                  |   3 +
 tests/test_calibration.py                     |   4 +
 tests/test_findings.py                        |  32 ++
 tests/test_harness.py                         | 104 ++++
 tests/test_mapping.py                         | 215 +++++++++
 tests/test_rank.py                            |  23 +
 16 files changed, 1339 insertions(+), 53 deletions(-)
 create mode 100644 src/openultrasast/harness.py
 create mode 100644 src/openultrasast/mapping.py
 create mode 100644 tests/test_harness.py
 create mode 100644 tests/test_mapping.py

diff --git a/.kiro/specs/openrouter-sast-harness/design.md b/.kiro/specs/openrouter-sast-harness/design.md
index c781749..48ab612 100644
--- a/.kiro/specs/openrouter-sast-harness/design.md
+++ b/.kiro/specs/openrouter-sast-harness/design.md
@@ -19,7 +19,7 @@ Conceptual lineage:
 repo path
   -> intake snapshot
   -> preprocess and static tags
-  -> graph and optional SARIF ingest
+  -> graph, SARIF ingest, and entry-point reachability mapping
   -> embeddings and retrieval index
   -> file/function ranking
   -> tiered hunter pool
@@ -212,6 +212,7 @@ Primary mapping sources:
 ```text
 Semgrep: syntactic and lightweight semantic patterns, custom rule iteration, variant seeds
 CodeQL: interprocedural dataflow, taint paths, structural vulnerability queries
+Entry-point analysis: routes, CLI/process boundaries, parser inputs, smart-contract state-changing entry points, callbacks, privileged surfaces
 Differential review: changed attack surfaces, regression-prone patches, new trust-boundary crossings
 Sharp-edge review: unsafe APIs, insecure defaults, confusing configuration, misuse-prone interfaces
 SARIF parsing: normalized evidence, rule metadata, locations, severity, fingerprints
@@ -226,10 +227,41 @@ taint_paths and source/sink pairs
 variant_search_seeds
 changed_surface records
 sharp_edge records
+entry_point records
+reachability_hints on FileTarget
 evidence candidates for verifier
 ```
 
-This layer should use Trail of Bits skills as task templates and checklists. For example, the CodeQL discipline supplies query/dataflow thinking, Semgrep supplies pattern and variant loops, differential review supplies changed-code blast-radius analysis, and sharp-edge review supplies API misuse mapping.
+This layer should use Trail of Bits skills as task templates and checklists. For example, the CodeQL discipline supplies query/dataflow thinking, Semgrep supplies pattern and variant loops, entry-point analysis supplies attacker-controlled surface and access classification, differential review supplies changed-code blast-radius analysis, and sharp-edge review supplies API misuse mapping.
+
+#### Entry-Point Reachability Mapping
+
+Entry-point mapping is a Phase 5A prerequisite for useful ranking and false-positive reduction. Pattern matches and static analyzer alerts are not enough; the harness must identify whether the affected function or symbol is plausibly reachable from attacker-controlled input.
+
+Initial entry-point kinds:
+
+```text
+route: HTTP/framework route handlers and request callbacks
+cli: process entry points, argparse/click handlers, argv consumers, main functions
+parser: parser or deserialization inputs that consume untrusted data
+fuzz: explicit fuzz entry points such as LLVMFuzzerTestOneInput
+smart_contract_state_change: externally callable non-view contract functions
+contract_callback: callbacks and hook implementations invoked by other contracts
+privileged_surface: auth boundaries, admin operations, filesystem/process/syscall boundaries
+```
+
+Initial access levels:
+
+```text
+public
+authenticated
+role-restricted
+contract-only/callback
+local-only
+review-required
+```
+
+Each entry-point record should include path, function name when known, start line, end line, name, kind, access level, trust boundary, access evidence, condition evidence, provenance, and rationale. Access level distinguishes unauthenticated/public from authenticated, role-restricted, local-only, callback-only, and review-required surfaces. Condition evidence captures feature flags, rollout toggles, pause guards, experiment gates, and runtime guard expressions. The preprocessor attaches matching records to `FileTarget.reachability_hints`. Findings use the source line to inherit only function-level entry points whose range contains the finding. The ranker uses these hints to raise file review priority for files containing public routes, parser/fuzz inputs, authenticated surfaces, contract callbacks, and role-restricted privileged surfaces. A finding outside a reachable function should remain eligible for review, but should not inherit the entry point's priority without call-graph evidence.
 
 ### Ranker
 
diff --git a/.kiro/specs/openrouter-sast-harness/requirements.md b/.kiro/specs/openrouter-sast-harness/requirements.md
index 21bea86..834dfaf 100644
--- a/.kiro/specs/openrouter-sast-harness/requirements.md
+++ b/.kiro/specs/openrouter-sast-harness/requirements.md
@@ -16,6 +16,30 @@ The project shall use Clearwing only as an implementation oracle. It shall not i
 6. Integrate Trail of Bits skills as scoped security expertise selected by language, framework, vulnerability class, mapping need, or verification need.
 7. Emit machine-readable artifacts suitable for CI, triage, and later training or harness evolution.
 
+## Completeness Model
+
+Implementation progress shall be reported against functional harness capability, not raw checklist percentage. A task may be complete as scaffold while the product capability remains incomplete.
+
+Completion levels:
+
+1. `scaffold`: schemas, CLI shape, local artifacts, tests, or isolated primitives exist.
+2. `usable`: the feature runs end-to-end for a narrow local fixture without manual wiring.
+3. `integrated`: the feature participates in the scan harness with traces, evidence, reports, and degradation handling.
+4. `production`: the feature is robust enough for real repositories, CI budgets, failure recovery, and analyst audit.
+
+Progress reports shall identify both checklist progress and functional completeness. The project shall not describe OpenUltraSAST as a working harness until the `usable_harness_mvp` gate is satisfied.
+
+Functional gates:
+
+1. `scaffold_quick_scan`: quick scan can enumerate files, rank targets, emit basic static findings, and write JSON/Markdown artifacts. This gate does not constitute a security harness.
+2. `usable_harness_mvp`: harness runtime, event tracing, static mapping ingestion, evidence ladder enforcement, independent verification, and JSON/Markdown/SARIF reporting work together on fixture repositories.
+3. `standard_security_harness`: standard mode runs ranked hunters with retrieval packages, selected security skills, independent verification, false-positive learning, and auditable traces.
+4. `sandboxed_dynamic_harness`: Docker sandbox, scoped dynamic probes, sanitizer/build artifacts, and evidence transitions work without hidden network or host mutation.
+5. `fix_validation_harness`: patch proposals, adversarial fix audit, differential/static delta checks, and sandbox validation are integrated and cannot mark patches validated without passing evidence.
+6. `opencode_product`: MCP tools, opencode commands/skills, fusion deepening, CI-ready reports, cost controls, and hardening are complete enough for routine opencode-driven use.
+
+The current implementation state shall be documented as `scaffold_quick_scan` until later gates are explicitly verified.
+
 ## Non-Goals
 
 1. Do not make network pentesting the core workflow.
@@ -46,9 +70,13 @@ The project shall use Clearwing only as an implementation oracle. It shall not i
 
 1. The system shall treat Semgrep, CodeQL, differential review, and sharp-edge/API review as mapping disciplines, not just optional external scanners.
 2. The system shall ingest Semgrep and CodeQL output as SARIF and normalize it into static hints, evidence candidates, and variant-search seeds.
-3. The system shall support differential analysis between a base and head revision to identify changed attack surfaces, newly reachable code, and regression-prone fixes.
-4. The system shall support sharp-edge mapping for APIs, configuration, defaults, cryptographic interfaces, and dangerous framework patterns that create predictable misuse.
-5. The system shall preserve analyzer provenance so each finding can cite which static tool, rule, diff, or skill-derived checklist contributed evidence.
+3. The system shall support function-level entry-point and reachability analysis to identify attacker-controlled inputs, externally callable functions, public routes, CLI/process boundaries, parser entry points, smart-contract state-changing entry points, callbacks, and privileged/admin-only surfaces.
+4. The system shall classify entry points by access level and trust boundary, including public/unrestricted, authenticated, role-restricted, contract-only/callback, local-only, and unknown/review-required.
+5. The system shall preserve condition evidence for entry points, including feature flags, rollout toggles, pause guards, experiment gates, and runtime guard expressions that affect whether a threat actor can reach the code.
+6. The system shall use function-level reachability modeling to prioritize candidate bugs inside reachable entry-point functions and to demote pattern-only findings that lack a plausible attacker-controlled path.
+6. The system shall support differential analysis between a base and head revision to identify changed attack surfaces, newly reachable code, and regression-prone fixes.
+7. The system shall support sharp-edge mapping for APIs, configuration, defaults, cryptographic interfaces, and dangerous framework patterns that create predictable misuse.
+8. The system shall preserve analyzer provenance so each finding can cite which static tool, rule, diff, entry-point analysis, or skill-derived checklist contributed evidence.
 
 ### OpenRouter Embeddings And Retrieval
 
@@ -65,7 +93,7 @@ The project shall use Clearwing only as an implementation oracle. It shall not i
 1. The system shall score candidate files on at least three axes: surface, influence, and reachability.
 2. The system shall compute a composite priority using an explicit formula, initially `surface * 0.5 + influence * 0.2 + reachability * 0.3`.
 3. The system shall keep low-surface high-influence files eligible for analysis.
-4. The system shall combine static heuristics, dependency graph data, embeddings, and OpenRouter LLM scoring.
+4. The system shall combine static heuristics, dependency graph data, entry-point reachability, embeddings, and OpenRouter LLM scoring.
 5. The system shall record score rationales and model identity.
 6. The system shall measure ranking quality with fixture repositories and prior scan outcomes, including false-positive rate by tier and missed-true-positive rate where ground truth exists.
 7. The system shall feed verifier outcomes back into ranking calibration so repeated false-positive patterns lose priority over time.
@@ -165,9 +193,11 @@ The project shall use Clearwing only as an implementation oracle. It shall not i
 
 ## Acceptance Criteria
 
-1. A user can run a local repository scan in quick mode and receive ranked static/LLM findings with JSON and Markdown reports.
-2. A user can run standard mode and receive adversarially verified findings with explicit evidence levels.
-3. A user can run deep mode on a small C/C++ parser target and receive sandboxed build/fuzz artifacts when the target is fuzzable.
-4. A finding cannot be marked verified unless it reaches at least `static_corroboration`.
-5. A patch cannot be marked validated unless a sandboxed validation command passes.
-6. A scan manifest records repository snapshot, harness config, OpenRouter model IDs, embedding model, prompt hashes, and artifact paths.
+1. `scaffold_quick_scan`: A user can run a local repository scan in quick mode and receive ranked static findings with JSON and Markdown reports. This may use heuristics and static patterns only.
+2. `usable_harness_mvp`: A user can run quick or standard fixture scans through the harness runtime and inspect event traces, normalized static mapping evidence, verifier decisions, and report artifacts.
+3. `standard_security_harness`: A user can run standard mode and receive adversarially verified findings with explicit evidence levels, selected retrieval context, skill routing, and false-positive learning records.
+4. `sandboxed_dynamic_harness`: A user can run deep mode on a small C/C++ parser target and receive sandboxed build/fuzz artifacts when the target is fuzzable.
+5. `fix_validation_harness`: A patch cannot be marked validated unless a sandboxed validation command passes and the fix audit has no accepted blocking findings.
+6. `opencode_product`: A user can drive scans, triage, evidence inspection, report export, fusion, and patch proposal through narrow MCP/OpenCode interfaces without exposing arbitrary shell execution.
+7. A finding cannot be marked verified unless it reaches at least `static_corroboration` through enforced evidence transitions.
+8. A scan manifest records repository snapshot, harness config, OpenRouter model IDs, embedding model, prompt hashes, processor versions, evidence transitions, and artifact paths.
diff --git a/.kiro/specs/openrouter-sast-harness/tasks.md b/.kiro/specs/openrouter-sast-harness/tasks.md
index 4213e93..8dd0f90 100644
--- a/.kiro/specs/openrouter-sast-harness/tasks.md
+++ b/.kiro/specs/openrouter-sast-harness/tasks.md
@@ -1,6 +1,19 @@
 # Tasks: OpenRouter SAST Harness
 
-## Phase 0: Project Skeleton
+## Completion Gates
+
+Checklist completion is not product completion. Each phase now carries a functional gate label:
+
+- `scaffold`: isolated primitives, schemas, local artifacts, or tests exist.
+- `usable`: the feature works end-to-end for fixtures without manual wiring.
+- `integrated`: the feature participates in the harness runtime, traces, evidence, and reports.
+- `production`: the feature is robust for real repositories, CI budgets, and analyst audit.
+
+Current verified gate: `scaffold_quick_scan`.
+
+Do not report OpenUltraSAST as a working security harness until `usable_harness_mvp` is checked.
+
+## Phase 0: Project Skeleton (`scaffold`, complete)
 
 - [x] Create Python package skeleton with CLI entry point `ousast`.
 - [x] Add config loading for `openultrasast.toml` and environment variables.
@@ -58,26 +71,51 @@
 - [x] Feed rejected findings into retrieval filter adjustment and prompt constraints.
 - [x] Add tests proving scoped false-positive demotion does not globally suppress the vulnerability class.
 
-## Phase 5: Harness Runtime
+## Gate: `scaffold_quick_scan`
 
-- [ ] Define harness event model.
-- [ ] Define processor interface and contract checks.
-- [ ] Implement strict and warn modes.
-- [ ] Add trace writer for events and processor state changes.
-- [ ] Serialize harness config, prompt hashes, processor versions, and model roles.
-- [ ] Add tests for processor contract violations.
+- [x] Quick scan can enumerate files, rank targets, emit basic static findings, and write JSON/Markdown artifacts.
+- [x] Quick scan is uv-managed and covered by unit tests.
+- [ ] Quick scan includes real LLM hunter execution. This is intentionally deferred to standard mode.
+- [ ] Quick scan includes SARIF output. This is intentionally deferred to Phase 9.
+- [ ] Quick scan includes enforced verifier decisions. This is intentionally deferred to Phase 6.
+
+## Gate: `usable_harness_mvp`
+
+- [x] Harness runtime drives scan stages instead of direct CLI orchestration.
+- [x] Event traces are emitted for every scan stage.
+- [x] Static mapping ingestion feeds FileTarget hints and verifier evidence candidates.
+- [x] Entry-point reachability feeds FileTarget reachability hints and ranking priority.
+- [ ] Evidence ladder state machine prevents invalid verified statuses.
+- [ ] Independent verifier runs on fixture findings without hunter reasoning.
+- [ ] JSON, Markdown, SARIF, and manifest artifacts share finding IDs and evidence references.
+- [ ] A fixture scan demonstrates one accepted static-corroborated finding and one rejected false positive.
+
+## Phase 5: Harness Runtime (`usable_harness_mvp` prerequisite)
+
+- [x] Define harness event model.
+- [x] Define processor interface and contract checks.
+- [x] Implement strict and warn modes.
+- [x] Add trace writer for events and processor state changes.
+- [x] Serialize harness config, prompt hashes, processor versions, and model roles.
+- [x] Add tests for processor contract violations.
 
 ## Phase 5A: Static Mapping Disciplines
 
-- [ ] Add SARIF ingestion for Semgrep and CodeQL outputs.
-- [ ] Normalize SARIF rules, locations, severity, fingerprints, and provenance into static hints.
-- [ ] Add Semgrep mapping task records for pattern, variant, and rule-authoring loops.
-- [ ] Add CodeQL mapping task records for source, sink, sanitizer, and path explanations.
-- [ ] Add differential mapping records for changed files, changed functions, and changed trust-boundary crossings.
-- [ ] Add sharp-edge records for insecure defaults, dangerous APIs, confusing config, and misuse-prone interfaces.
-- [ ] Add tests for mapping analyzer output to `FileTarget` hints and verifier evidence candidates.
+- [x] Add SARIF ingestion for Semgrep and CodeQL outputs.
+- [x] Normalize SARIF rules, locations, severity, fingerprints, and provenance into static hints.
+- [x] Add Semgrep mapping task records for pattern, variant, and rule-authoring loops.
+- [x] Add CodeQL mapping task records for source, sink, sanitizer, and path explanations.
+- [x] Add entry-point mapping records for routes, CLI/process boundaries, parser entry points, smart-contract state-changing entry points, callbacks, and privileged surfaces.
+- [x] Classify entry points by access level: public, authenticated, role-restricted, contract-only/callback, local-only, or review-required.
+- [x] Map entry points to function-level source ranges and access evidence.
+- [x] Preserve conditional reachability evidence for feature flags, rollout toggles, pause guards, and runtime guard expressions.
+- [x] Attach reachability hints to FileTargets and ranking inputs.
+- [x] Annotate findings with function-level reachability status before prioritizing fixes.
+- [x] Add differential mapping records for changed files, changed functions, and changed trust-boundary crossings.
+- [x] Add sharp-edge records for insecure defaults, dangerous APIs, confusing config, and misuse-prone interfaces.
+- [x] Add tests for mapping analyzer output to `FileTarget` hints and verifier evidence candidates.
 
-## Phase 6: Verification
+## Phase 6: Verification (`usable_harness_mvp` prerequisite)
 
 - [ ] Define evidence ladder state machine.
 - [ ] Implement independent verifier prompt and schema.
@@ -86,7 +124,14 @@
 - [ ] Enforce that verified report status requires at least `static_corroboration`.
 - [ ] Add tests for invalid evidence transitions.
 
-## Phase 7: Docker Sandbox
+## Gate: `standard_security_harness`
+
+- [ ] Standard mode uses ranked hunter scheduling, retrieval packages, selected skill snippets, and independent verification.
+- [ ] Every hunter trajectory is persisted and linked to finding artifacts.
+- [ ] False-positive learning updates ranking calibration from verifier outcomes.
+- [ ] Fusion can be triggered for high-impact, contradictory, or difficult findings.
+
+## Phase 7: Docker Sandbox (`sandboxed_dynamic_harness` prerequisite)
 
 - [ ] Implement Docker availability probe.
 - [ ] Implement no-network read-only workspace sandbox runner.
@@ -97,7 +142,7 @@
 - [ ] Add sanitizer build environment for C/C++ fixtures.
 - [ ] Add tests that verify workspace remains read-only.
 
-## Phase 7A: Dynamic Evidence
+## Phase 7A: Dynamic Evidence (`sandboxed_dynamic_harness` prerequisite)
 
 - [ ] Implement dynamic analysis configuration with `enabled = false` by default.
 - [ ] Implement service startup command records with declared ports and timeouts.
@@ -106,7 +151,7 @@
 - [ ] Tie dynamic artifacts to evidence transitions without allowing model-only upgrades.
 - [ ] Add tests that reject undeclared network targets.
 
-## Phase 8: Standard Mode Hunter Pool
+## Phase 8: Standard Mode Hunter Pool (`standard_security_harness` prerequisite)
 
 - [ ] Implement tier assignment by priority.
 - [ ] Implement budget allocation and per-target limits.
@@ -116,7 +161,7 @@
 - [ ] Integrate verifier after hunter output.
 - [ ] Add fixture scan demonstrating at least one verified static-corroborated finding.
 
-## Phase 9: SARIF And CI Output
+## Phase 9: SARIF And CI Output (`usable_harness_mvp` prerequisite)
 
 - [ ] Implement SARIF report writer.
 - [ ] Implement machine-readable manifest.
@@ -124,7 +169,13 @@
 - [ ] Add CI-friendly exit code policy.
 - [ ] Add tests for SARIF schema validity.
 
-## Phase 10: Deep Mode
+## Gate: `sandboxed_dynamic_harness`
+
+- [ ] Docker sandbox runs build/test/fuzz/PoC commands with no network by default.
+- [ ] Scoped dynamic probes require explicit declared targets and emit artifacts.
+- [ ] Sanitizer/fuzz artifacts can advance evidence only through valid transitions.
+
+## Phase 10: Deep Mode (`sandboxed_dynamic_harness` prerequisite)
 
 - [ ] Implement fuzzable target detection beyond existing fuzz entry points.
 - [ ] Implement fuzz harness generation prompt and schema.
@@ -134,7 +185,13 @@
 - [ ] Advance findings to `crash_reproduced` only with artifact evidence.
 - [ ] Add a small C/C++ parser fixture with reproducible sanitizer crash.
 
-## Phase 11: Patch Oracle
+## Gate: `fix_validation_harness`
+
+- [ ] Patch proposals are generated as diffs and not silently applied.
+- [ ] Fix lifecycle records intake, plan, implementation, adversarial review, reconciliation, verification, and ready states.
+- [ ] Patch validation requires sandboxed checks and accepted adversarial findings must be resolved.
+
+## Phase 11: Patch Oracle (`fix_validation_harness` prerequisite)
 
 - [ ] Implement patch proposal schema.
 - [ ] Implement OpenUltraCode-style fix lifecycle states: intake, plan, implement, adversarial_review, reconcile, verify, ready.
@@ -147,7 +204,13 @@
 - [ ] Advance findings to `patch_validated` only after validation passes.
 - [ ] Add tests that reject patches changing unrelated files.
 
-## Phase 12: MCP And OpenCode Commands
+## Gate: `opencode_product`
+
+- [ ] Narrow MCP tools cover scan, status, findings, evidence, artifacts, patch proposal, and report export.
+- [ ] OpenCode workflows can drive scan, triage, fusion, and patch audit without arbitrary shell exposure.
+- [ ] Cost controls, retries, redaction, docs, examples, and CI gates are in place.
+
+## Phase 12: MCP And OpenCode Commands (`opencode_product` prerequisite)
 
 - [x] Add project-local opencode skills for scan, triage, fix audit, and implementation steering.
 - [ ] Implement narrow MCP server.
@@ -156,7 +219,7 @@
 - [ ] Ensure MCP tools never expose arbitrary shell execution.
 - [ ] Add integration test for MCP tool listing and scan status.
 
-## Phase 13: Fusion
+## Phase 13: Fusion (`standard_security_harness` and `opencode_product` prerequisite)
 
 - [ ] Implement fusion trigger policy.
 - [ ] Trigger fusion whenever a task requires deeper reasoning, including difficult findings, contradictory evidence, high-impact decisions, and risky fixes.
@@ -167,7 +230,7 @@
 - [ ] Record model IDs, degradations, warnings, and final decision source.
 - [ ] Add tests for fusion output validation.
 
-## Phase 14: Skill Index
+## Phase 14: Skill Index (`standard_security_harness` prerequisite)
 
 - [ ] Build Trail of Bits skill descriptor inventory.
 - [ ] Chunk selected skill operating guidance into retrieval namespace.
@@ -175,7 +238,7 @@
 - [ ] Add prompt budget caps for skill snippets.
 - [ ] Add tests for routing C parser targets to memory/fuzzing skills and crypto targets to constant-time/vector skills.
 
-## Phase 15: Hardening And Release Readiness
+## Phase 15: Hardening And Release Readiness (`production` prerequisite)
 
 - [ ] Add cost budget enforcement.
 - [ ] Add model call retry and backoff policies.
diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py
index 84bf368..2ebe3bc 100644
--- a/src/openultrasast/cli.py
+++ b/src/openultrasast/cli.py
@@ -6,8 +6,10 @@ from pathlib import Path
 
 from .config import load_config
 from .findings import quick_scan_findings, write_findings
+from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config
 from .index import build_code_chunks
-from .preprocess import preprocess_repository
+from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints
+from .preprocess import preprocess_repository, write_preprocess_artifact
 from .rank import rank_targets, write_rankings
 from .reports import write_markdown_report
 from .run import create_scan_run
@@ -43,12 +45,29 @@ def _scan(path: Path, config_path: Path, mode: str) -> int:
 
     config = load_config(config_path if config_path.exists() else None)
     run = create_scan_run(path, config)
-    _, targets = preprocess_repository(run.target, run.root / "preprocess" / "file_targets.json")
-    rankings = rank_targets(targets)
+    runtime = HarnessRuntime(
+        scan_id=run.scan_id,
+        config=config,
+        trace_writer=HarnessTraceWriter(run.root / "trace" / "events.jsonl"),
+    )
+    write_harness_config(config=config, processors=[], contract_mode="strict", path=run.root / "harness.json")
+    runtime.start(mode=mode, target=run.target)
+    static_hints = runtime.run_stage("static_mapping", lambda: _load_static_hints(config.static_analysis.sarif_paths))
+    write_static_hints(static_hints, run.root / "mapping" / "static_hints.json")
+    snapshot, targets = runtime.run_stage(
+        "preprocess",
+        lambda: preprocess_repository(run.target, run.root / "preprocess" / "file_targets.json", static_hints),
+    )
+    entry_points = runtime.run_stage("entry_point_mapping", lambda: analyze_entry_points(run.target, targets))
+    write_entry_points(entry_points, run.root / "mapping" / "entry_points.json")
+    targets = attach_reachability_hints(targets, entry_points)
+    write_preprocess_artifact(snapshot, targets, run.root / "preprocess" / "file_targets.json")
+    rankings = runtime.run_stage("rank", lambda: rank_targets(targets))
     write_rankings(rankings, run.root / "rank" / "ranking.json")
-    findings = quick_scan_findings(run.target, targets, rankings)
+    findings = runtime.run_stage("quick_findings", lambda: quick_scan_findings(run.target, targets, rankings))
     write_findings(findings, run.root / "findings.json")
-    write_markdown_report(findings, run.root / "report.md")
+    runtime.run_stage("report", lambda: write_markdown_report(findings, run.root / "report.md"))
+    runtime.finish(status="succeeded")
     print(f"scan_id={run.scan_id}")
     print(f"run_dir={run.root}")
     print(f"file_targets={len(targets)}")
@@ -57,6 +76,15 @@ def _scan(path: Path, config_path: Path, mode: str) -> int:
     return 0
 
 
+def _load_static_hints(sarif_paths: tuple[str, ...]) -> list[object]:
+    hints: list[object] = []
+    for sarif_path in sarif_paths:
+        path = Path(sarif_path)
+        if path.exists():
+            hints.extend(ingest_sarif(path))
+    return hints
+
+
 def _index(path: Path, config_path: Path, chunk_lines: int) -> int:
     if not path.exists() or not path.is_dir():
         raise SystemExit(f"index path is not a directory: {path}")
diff --git a/src/openultrasast/config.py b/src/openultrasast/config.py
index 2721238..9c517f8 100644
--- a/src/openultrasast/config.py
+++ b/src/openultrasast/config.py
@@ -46,6 +46,11 @@ class EvidenceConfig:
     minimum_patch: str = "root_cause_explained"
 
 
+@dataclass(frozen=True)
+class StaticAnalysisConfig:
+    sarif_paths: tuple[str, ...] = ()
+
+
 @dataclass(frozen=True)
 class ResolvedConfig:
     models: ModelConfig = ModelConfig()
@@ -53,6 +58,7 @@ class ResolvedConfig:
     sandbox: SandboxConfig = SandboxConfig()
     dynamic: DynamicConfig = DynamicConfig()
     evidence: EvidenceConfig = EvidenceConfig()
+    static_analysis: StaticAnalysisConfig = StaticAnalysisConfig()
     runs_dir: str = ".openultrasast/runs"
 
 
@@ -68,6 +74,7 @@ def load_config(config_path: Path | None = None) -> ResolvedConfig:
         sandbox=_load_sandbox(data.get("sandbox", {})),
         dynamic=_load_dynamic(data.get("dynamic", {})),
         evidence=_load_evidence(data.get("evidence", {})),
+        static_analysis=_load_static_analysis(data.get("static_analysis", {})),
         runs_dir=os.environ.get("OPENULTRASAST_RUNS_DIR", ".openultrasast/runs"),
     )
 
@@ -126,3 +133,11 @@ def _load_evidence(value: object) -> EvidenceConfig:
         minimum_exploit=str(data.get("minimum_exploit", "crash_reproduced")),
         minimum_patch=str(data.get("minimum_patch", "root_cause_explained")),
     )
+
+
+def _load_static_analysis(value: object) -> StaticAnalysisConfig:
+    data = _section(value)
+    paths = data.get("sarif_paths", [])
+    if not isinstance(paths, list):
+        paths = []
+    return StaticAnalysisConfig(sarif_paths=tuple(str(item) for item in paths))
diff --git a/src/openultrasast/findings.py b/src/openultrasast/findings.py
index beed6f3..6cb3e38 100644
--- a/src/openultrasast/findings.py
+++ b/src/openultrasast/findings.py
@@ -19,6 +19,10 @@ class StaticFinding:
     evidence_level: str
     rationale: str
     line: int | None
+    function_name: str | None
+    reachability_status: str
+    reachability_evidence: list[dict[str, object]]
+    reachability_conditions: list[str]
     tags: list[str]
     ranking_priority: float
 
@@ -70,11 +74,9 @@ def quick_scan_findings(root: Path, targets: list[FileTarget], rankings: list[Ra
     for target in targets:
         text = _read_text(root / target.path)
         for rule in PATTERN_RULES:
-            match = rule.pattern.search(text)
-            if match is None:
-                continue
             ranking = ranking_by_path.get(target.path)
-            findings.append(_finding_from_match(target, rule, text, match.start(), ranking))
+            for match in rule.pattern.finditer(text):
+                findings.append(_finding_from_match(target, rule, text, match.start(), ranking))
     return sorted(findings, key=lambda item: (_severity_sort(item.severity), -item.ranking_priority, item.path))
 
 
@@ -103,6 +105,8 @@ def _finding_from_match(
     ranking: RankingScore | None,
 ) -> StaticFinding:
     line = text.count("\n", 0, offset) + 1
+    reachability_status, function_name, reachability_evidence = _reachability_for_line(target, line)
+    priority = ranking.priority if ranking is not None else 1.0
     return StaticFinding(
         finding_id=f"{rule.rule_id}:{target.path}:{line}",
         path=target.path,
@@ -112,11 +116,62 @@ def _finding_from_match(
         evidence_level="static_corroboration",
         rationale=f"Static pattern {rule.rule_id} matched line {line}; manual verification still required.",
         line=line,
+        function_name=function_name,
+        reachability_status=reachability_status,
+        reachability_evidence=reachability_evidence,
+        reachability_conditions=_reachability_conditions(reachability_evidence),
         tags=sorted(set(target.tags) | set(rule.tags)),
-        ranking_priority=ranking.priority if ranking is not None else 1.0,
+        ranking_priority=_finding_priority(priority, reachability_status),
     )
 
 
+def _reachability_for_line(target: FileTarget, line: int) -> tuple[str, str | None, list[dict[str, object]]]:
+    hints = target.reachability_hints
+    concrete = [hint for hint in hints if isinstance(hint.get("line"), int)]
+    matching = [hint for hint in concrete if _line_in_hint(line, hint)]
+    if matching:
+        return "reachable", _function_name(matching), matching
+    if concrete:
+        return "unknown", None, []
+    inferred = [hint for hint in hints if hint.get("line") is None]
+    if inferred:
+        return "inferred-file-surface", None, inferred
+    return "unknown", None, []
+
+
+def _line_in_hint(line: int, hint: dict[str, object]) -> bool:
+    start = hint.get("line")
+    end = hint.get("end_line")
+    if not isinstance(start, int):
+        return False
+    if not isinstance(end, int):
+        end = start
+    return start <= line <= end
+
+
+def _function_name(hints: list[dict[str, object]]) -> str | None:
+    for hint in hints:
+        function_name = hint.get("function_name")
+        if isinstance(function_name, str) and function_name:
+            return function_name
+    return None
+
+
+def _reachability_conditions(hints: list[dict[str, object]]) -> list[str]:
+    conditions: list[str] = []
+    for hint in hints:
+        value = hint.get("conditions")
+        if isinstance(value, list):
+            conditions.extend(str(item) for item in value if item)
+    return sorted(set(conditions))
+
+
+def _finding_priority(priority: float, reachability_status: str) -> float:
+    if reachability_status == "unknown":
+        return min(priority, 1.5)
+    return priority
+
+
 def _read_text(path: Path) -> str:
     try:
         return path.read_text(errors="ignore")
diff --git a/src/openultrasast/harness.py b/src/openultrasast/harness.py
new file mode 100644
index 0000000..4b88242
--- /dev/null
+++ b/src/openultrasast/harness.py
@@ -0,0 +1,187 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import asdict, dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, Callable, Literal, Protocol
+
+from .config import ResolvedConfig
+
+EventType = Literal[
+    "task_start",
+    "stage_start",
+    "before_model",
+    "after_model",
+    "before_tool",
+    "after_tool",
+    "stage_end",
+    "task_end",
+]
+
+ContractMode = Literal["strict", "warn"]
+
+
+class HarnessContractError(ValueError):
+    """Raised when a processor violates its declared state contract."""
+
+
+@dataclass(frozen=True)
+class HarnessEvent:
+    event_type: EventType
+    scan_id: str
+    stage: str | None
+    timestamp: str
+    payload: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class ProcessorSpec:
+    name: str
+    version: str
+    handles: tuple[EventType, ...]
+    reads: tuple[str, ...] = ()
+    writes: tuple[str, ...] = ()
+
+
+class Processor(Protocol):
+    spec: ProcessorSpec
+
+    def handle(self, event: HarnessEvent, state: dict[str, Any]) -> dict[str, Any] | None:
+        raise NotImplementedError
+
+
+@dataclass(frozen=True)
+class SerializedHarnessConfig:
+    contract_mode: ContractMode
+    model_roles: dict[str, str | None]
+    processor_versions: dict[str, str]
+    prompt_hashes: dict[str, str]
+
+
+class HarnessTraceWriter:
+    def __init__(self, path: Path) -> None:
+        self.path = path
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+
+    def append(self, event: HarnessEvent) -> None:
+        with self.path.open("a") as handle:
+            handle.write(json.dumps(asdict(event), sort_keys=True) + "\n")
+
+
+class HarnessRuntime:
+    def __init__(
+        self,
+        *,
+        scan_id: str,
+        config: ResolvedConfig,
+        trace_writer: HarnessTraceWriter,
+        processors: list[Processor] | None = None,
+        contract_mode: ContractMode = "strict",
+    ) -> None:
+        self.scan_id = scan_id
+        self.config = config
+        self.trace_writer = trace_writer
+        self.processors = processors or []
+        self.contract_mode = contract_mode
+        self.state: dict[str, Any] = {"degradations": []}
+
+    def start(self, *, mode: str, target: Path) -> None:
+        self.emit("task_start", None, {"mode": mode, "target": str(target)})
+
+    def finish(self, *, status: str) -> None:
+        self.emit("task_end", None, {"status": status})
+
+    def run_stage(self, name: str, callback: Callable[[], Any]) -> Any:
+        self.emit("stage_start", name, {})
+        try:
+            result = callback()
+        except Exception as exc:
+            self.emit("stage_end", name, {"status": "failed", "error": f"{type(exc).__name__}: {exc}"})
+            raise
+        self.emit("stage_end", name, {"status": "succeeded"})
+        return result
+
+    def emit(self, event_type: EventType, stage: str | None, payload: dict[str, Any]) -> None:
+        event = HarnessEvent(
+            event_type=event_type,
+            scan_id=self.scan_id,
+            stage=stage,
+            timestamp=datetime.now(UTC).isoformat(),
+            payload=payload,
+        )
+        self.trace_writer.append(event)
+        for processor in self.processors:
+            if event_type in processor.spec.handles:
+                self._run_processor(processor, event)
+
+    def _run_processor(self, processor: Processor, event: HarnessEvent) -> None:
+        before = set(self.state)
+        missing_reads = [key for key in processor.spec.reads if key not in self.state]
+        if missing_reads:
+            self._contract_violation(processor.spec.name, f"missing reads: {', '.join(missing_reads)}")
+            return
+
+        updates = processor.handle(event, dict(self.state)) or {}
+        if not isinstance(updates, dict):
+            self._contract_violation(processor.spec.name, "processor returned a non-dict update")
+            return
+        undeclared = sorted(set(updates) - set(processor.spec.writes))
+        if undeclared:
+            self._contract_violation(processor.spec.name, f"undeclared writes: {', '.join(undeclared)}")
+            return
+        self.state.update(updates)
+        changed = sorted(set(self.state) - before | set(updates))
+        self.trace_writer.append(
+            HarnessEvent(
+                event_type=event.event_type,
+                scan_id=self.scan_id,
+                stage=event.stage,
+                timestamp=datetime.now(UTC).isoformat(),
+                payload={"processor": processor.spec.name, "changed_state": changed},
+            )
+        )
+
+    def _contract_violation(self, processor_name: str, reason: str) -> None:
+        message = f"processor {processor_name} contract violation: {reason}"
+        if self.contract_mode == "strict":
+            raise HarnessContractError(message)
+        self.state.setdefault("degradations", []).append(message)
+        self.trace_writer.append(
+            HarnessEvent(
+                event_type="stage_end",
+                scan_id=self.scan_id,
+                stage="processor_contract",
+                timestamp=datetime.now(UTC).isoformat(),
+                payload={"status": "degraded", "reason": message},
+            )
+        )
+
+
+def write_harness_config(
+    *,
+    config: ResolvedConfig,
+    processors: list[Processor],
+    contract_mode: ContractMode,
+    path: Path,
+    prompts: dict[str, str] | None = None,
+) -> None:
+    prompts = prompts or {}
+    serialized = SerializedHarnessConfig(
+        contract_mode=contract_mode,
+        model_roles={
+            "ranker": config.models.ranker,
+            "hunter": config.models.hunter,
+            "verifier": config.models.verifier,
+            "patcher": config.models.patcher,
+        },
+        processor_versions={processor.spec.name: processor.spec.version for processor in processors},
+        prompt_hashes={name: _hash_text(prompt) for name, prompt in sorted(prompts.items())},
+    )
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(asdict(serialized), indent=2, sort_keys=True) + "\n")
+
+
+def _hash_text(text: str) -> str:
+    return hashlib.sha256(text.encode()).hexdigest()
diff --git a/src/openultrasast/mapping.py b/src/openultrasast/mapping.py
new file mode 100644
index 0000000..75b6e19
--- /dev/null
+++ b/src/openultrasast/mapping.py
@@ -0,0 +1,455 @@
+from __future__ import annotations
+
+import ast
+import json
+from dataclasses import asdict, dataclass, replace
+from pathlib import Path
+from typing import Any, Literal
+
+from .preprocess import FileTarget
+
+MappingKind = Literal["semgrep", "codeql", "differential", "sharp_edge"]
+AccessLevel = Literal[
+    "public",
+    "authenticated",
+    "role-restricted",
+    "contract-only/callback",
+    "local-only",
+    "review-required",
+]
+
+
+@dataclass(frozen=True)
+class StaticHint:
+    analyzer: str
+    rule_id: str
+    title: str
+    severity: str
+    path: str
+    start_line: int | None
+    message: str
+    fingerprint: str | None
+    provenance: str
+    evidence_candidate: bool = True
+
+
+@dataclass(frozen=True)
+class MappingTaskRecord:
+    kind: MappingKind
+    path: str
+    rule_id: str
+    task: str
+    rationale: str
+
+
+@dataclass(frozen=True)
+class DifferentialMappingRecord:
+    path: str
+    change_kind: str
+    trust_boundary: str | None
+    rationale: str
+
+
+@dataclass(frozen=True)
+class SharpEdgeRecord:
+    path: str
+    category: str
+    surface: str
+    rationale: str
+
+
+@dataclass(frozen=True)
+class EntryPointRecord:
+    path: str
+    line: int | None
+    end_line: int | None
+    function_name: str | None
+    name: str
+    kind: str
+    access_level: AccessLevel
+    trust_boundary: str
+    access_evidence: list[str]
+    conditions: list[str]
+    provenance: str
+    rationale: str
+
+
+def ingest_sarif(path: Path) -> list[StaticHint]:
+    payload = json.loads(path.read_text())
+    hints: list[StaticHint] = []
+    for run in _items(payload.get("runs")):
+        tool = run.get("tool") if isinstance(run, dict) else {}
+        driver = tool.get("driver") if isinstance(tool, dict) else {}
+        analyzer = _normalise_analyzer(str(driver.get("name", "sarif"))) if isinstance(driver, dict) else "sarif"
+        rules = _rules_by_id(driver.get("rules", []) if isinstance(driver, dict) else [])
+        for result in _items(run.get("results") if isinstance(run, dict) else []):
+            hint = _hint_from_result(result, analyzer, rules)
+            if hint is not None:
+                hints.append(hint)
+    return sorted(hints, key=lambda item: (item.path, item.start_line or 0, item.rule_id))
+
+
+def attach_static_hints(targets: list[FileTarget], hints: list[StaticHint]) -> list[FileTarget]:
+    hints_by_path: dict[str, list[dict[str, object]]] = {}
+    for hint in hints:
+        hints_by_path.setdefault(hint.path, []).append(asdict(hint))
+    return [replace(target, static_hints=hints_by_path.get(target.path, target.static_hints)) for target in targets]
+
+
+def analyze_entry_points(root: Path, targets: list[FileTarget]) -> list[EntryPointRecord]:
+    records: list[EntryPointRecord] = []
+    for target in targets:
+        path = root / target.path
+        text = path.read_text(errors="ignore")
+        records.extend(_language_entry_points(target, text))
+        records.extend(_tag_entry_points(target))
+    return sorted(records, key=lambda item: (item.path, item.line is None, item.line or 0, item.name))
+
+
+def attach_reachability_hints(targets: list[FileTarget], entry_points: list[EntryPointRecord]) -> list[FileTarget]:
+    hints_by_path: dict[str, list[dict[str, object]]] = {}
+    for entry_point in entry_points:
+        hints_by_path.setdefault(entry_point.path, []).append(asdict(entry_point))
+    return [replace(target, reachability_hints=hints_by_path.get(target.path, target.reachability_hints)) for target in targets]
+
+
+def verifier_evidence_candidates(hints: list[StaticHint], *, path: str | None = None) -> list[dict[str, object]]:
+    selected = [hint for hint in hints if hint.evidence_candidate and (path is None or hint.path == path)]
+    return [
+        {
+            "source": hint.analyzer,
+            "rule_id": hint.rule_id,
+            "path": hint.path,
+            "line": hint.start_line,
+            "severity": hint.severity,
+            "message": hint.message,
+            "provenance": hint.provenance,
+        }
+        for hint in selected
+    ]
+
+
+def semgrep_mapping_tasks(hints: list[StaticHint]) -> list[MappingTaskRecord]:
+    return [
+        MappingTaskRecord(
+            kind="semgrep",
+            path=hint.path,
+            rule_id=hint.rule_id,
+            task="pattern_variant_review",
+            rationale="Use Semgrep evidence to check variants and rule precision before verifier promotion.",
+        )
+        for hint in hints
+        if hint.analyzer == "semgrep"
+    ]
+
+
+def codeql_mapping_tasks(hints: list[StaticHint]) -> list[MappingTaskRecord]:
+    return [
+        MappingTaskRecord(
+            kind="codeql",
+            path=hint.path,
+            rule_id=hint.rule_id,
+            task="source_sink_sanitizer_path_review",
+            rationale="Use CodeQL evidence to inspect source, sink, sanitizer, and path explanation.",
+        )
+        for hint in hints
+        if hint.analyzer == "codeql"
+    ]
+
+
+def differential_mapping_record(path: str, *, change_kind: str, trust_boundary: str | None, rationale: str) -> DifferentialMappingRecord:
+    return DifferentialMappingRecord(path=path, change_kind=change_kind, trust_boundary=trust_boundary, rationale=rationale)
+
+
+def sharp_edge_record(path: str, *, category: str, surface: str, rationale: str) -> SharpEdgeRecord:
+    return SharpEdgeRecord(path=path, category=category, surface=surface, rationale=rationale)
+
+
+def write_static_hints(hints: list[StaticHint], path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps({"static_hints": [asdict(hint) for hint in hints]}, indent=2, sort_keys=True) + "\n")
+
+
+def write_entry_points(entry_points: list[EntryPointRecord], path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps({"entry_points": [asdict(entry_point) for entry_point in entry_points]}, indent=2, sort_keys=True) + "\n")
+
+
+def _language_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
+    if target.language == "python":
+        return _python_entry_points(target, text)
+    if target.language in {"javascript", "typescript"}:
+        return _js_entry_points(target, text)
+    if target.language in {"c", "cpp"}:
+        return _c_entry_points(target, text)
+    if target.language == "solidity":
+        return _solidity_entry_points(target, text)
+    return []
+
+
+def _python_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
+    records: list[EntryPointRecord] = []
+    try:
+        tree = ast.parse(text)
+    except SyntaxError:
+        tree = None
+    if tree is not None:
+        for node in ast.walk(tree):
+            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                decorators = [_safe_unparse(decorator) for decorator in node.decorator_list]
+                route = next((decorator for decorator in decorators if _looks_like_route_decorator(decorator)), None)
+                if route is not None:
+                    access, evidence = _python_route_access(decorators)
+                    conditions = _python_conditions(node, decorators)
+                    records.append(
+                        _entry(
+                            target,
+                            node.lineno,
+                            getattr(node, "end_lineno", node.lineno),
+                            node.name,
+                            node.name,
+                            "route",
+                            access,
+                            "http_request",
+                            evidence or [route],
+                            conditions,
+                        )
+                    )
+                if any("click.command" in decorator or ".command" in decorator for decorator in decorators):
+                    records.append(
+                        _entry(
+                            target,
+                            node.lineno,
+                            getattr(node, "end_lineno", node.lineno),
+                            node.name,
+                            node.name,
+                            "cli",
+                            "local-only",
+                            "process_invocation",
+                            decorators,
+                            _python_conditions(node, decorators),
+                        )
+                    )
+    for number, line in enumerate(text.splitlines(), start=1):
+        stripped = line.strip()
+        if stripped.startswith("if __name__") and "__main__" in stripped:
+            records.append(_entry(target, number, number, "__main__", "__main__", "cli", "local-only", "process_invocation", [stripped], []))
+        if "argparse." in stripped or "click.command" in stripped:
+            records.append(_entry(target, number, number, None, "cli_parser", "cli", "local-only", "process_invocation", [stripped], []))
+    return records
+
+
+def _js_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
+    records: list[EntryPointRecord] = []
+    route_tokens = (".get(", ".post(", ".put(", ".delete(", ".patch(", ".use(")
+    for number, line in enumerate(text.splitlines(), start=1):
+        stripped = line.strip()
+        if any(token in stripped for token in route_tokens) and any(prefix in stripped for prefix in ("app", "router", "server")):
+            records.append(_entry(target, number, number, _js_handler_name(stripped), "http_handler", "route", "public", "http_request", [stripped], _line_conditions(stripped)))
+        if "process.argv" in stripped:
+            records.append(_entry(target, number, number, None, "process.argv", "cli", "local-only", "process_invocation", [stripped], _line_conditions(stripped)))
+    return records
+
+
+def _c_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
+    records: list[EntryPointRecord] = []
+    for number, line in enumerate(text.splitlines(), start=1):
+        stripped = line.strip()
+        if " main(" in f" {stripped}" or stripped.startswith("main("):
+            records.append(_entry(target, number, number, "main", "main", "cli", "local-only", "process_invocation", [stripped], _line_conditions(stripped)))
+        if "LLVMFuzzerTestOneInput" in stripped:
+            records.append(_entry(target, number, number, "LLVMFuzzerTestOneInput", "LLVMFuzzerTestOneInput", "fuzz", "public", "fuzzer_input", [stripped], _line_conditions(stripped)))
+    return records
+
+
+def _solidity_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
+    records: list[EntryPointRecord] = []
+    for number, line in enumerate(text.splitlines(), start=1):
+        stripped = line.strip()
+        if not stripped.startswith("function ") or not any(token in stripped for token in (" public", " external")):
+            continue
+        if " view" in stripped or " pure" in stripped:
+            continue
+        evidence = [token for token in ("onlyOwner", "onlyRole", "requiresAuth") if token in stripped]
+        conditions = [token for token in ("whenNotPaused", "whenPaused", "featureEnabled") if token in stripped]
+        access = "role-restricted" if evidence else "public"
+        name = stripped.removeprefix("function ").split("(", 1)[0]
+        if "callback" in name.lower() or name.startswith("onERC"):
+            access = "contract-only/callback"
+            evidence.append("callback naming convention")
+        records.append(_entry(target, number, number, name, name, "smart_contract_state_change", access, "contract_call", evidence or ["external/public function"], conditions))
+    return records
+
+
+def _tag_entry_points(target: FileTarget) -> list[EntryPointRecord]:
+    records: list[EntryPointRecord] = []
+    tags = set(target.tags)
+    if "parser" in tags or "deserialization" in tags:
+        records.append(_entry(target, None, None, None, "parser_input", "parser", "public", "attacker_controlled_input", ["parser/deserialization file tag"], []))
+    if "network_entry" in tags:
+        records.append(_entry(target, None, None, None, "network_surface", "route", "public", "network_request", ["network entry file tag"], []))
+    if "auth_boundary" in tags:
+        records.append(_entry(target, None, None, None, "auth_boundary", "privileged_surface", "authenticated", "identity_boundary", ["auth boundary file tag"], []))
+    if "syscall_entry" in tags or "filesystem_entry" in tags:
+        records.append(_entry(target, None, None, None, "local_privileged_boundary", "privileged_surface", "local-only", "local_process", ["syscall/filesystem file tag"], []))
+    return records
+
+
+def _entry(
+    target: FileTarget,
+    line: int | None,
+    end_line: int | None,
+    function_name: str | None,
+    name: str,
+    kind: str,
+    access_level: AccessLevel,
+    trust_boundary: str,
+    access_evidence: list[str],
+    conditions: list[str],
+) -> EntryPointRecord:
+    return EntryPointRecord(
+        path=target.path,
+        line=line,
+        end_line=end_line,
+        function_name=function_name,
+        name=name,
+        kind=kind,
+        access_level=access_level,
+        trust_boundary=trust_boundary,
+        access_evidence=access_evidence,
+        conditions=conditions,
+        provenance=f"entrypoint:{kind}:{access_level}",
+        rationale=f"{kind} surface classified as {access_level} at {trust_boundary}",
+    )
+
+
+def _looks_like_route_decorator(line: str) -> bool:
+    return any(token in line for token in ("route", ".get", ".post", ".put", ".delete", ".patch"))
+
+
+def _safe_unparse(node: ast.AST) -> str:
+    try:
+        return ast.unparse(node)
+    except (AttributeError, ValueError):
+        return ""
+
+
+def _python_route_access(decorators: list[str]) -> tuple[AccessLevel, list[str]]:
+    role_tokens = ("role", "permission", "admin", "owner")
+    auth_tokens = ("login_required", "authenticated", "auth", "jwt_required")
+    access_decorators = [decorator for decorator in decorators if not _looks_like_route_decorator(decorator)]
+    evidence = [decorator for decorator in access_decorators if any(token in decorator.lower() for token in role_tokens + auth_tokens)]
+    lowered = "\n".join(evidence).lower()
+    if any(token in lowered for token in role_tokens):
+        return "role-restricted", evidence
+    if any(token in lowered for token in auth_tokens):
+        return "authenticated", evidence
+    return "public", evidence
+
+
+def _python_conditions(node: ast.FunctionDef | ast.AsyncFunctionDef, decorators: list[str]) -> list[str]:
+    conditions = [decorator for decorator in decorators if _looks_conditional(decorator)]
+    for child in ast.walk(node):
+        if isinstance(child, ast.If):
+            test = _safe_unparse(child.test)
+            if _looks_conditional(test):
+                conditions.append(test)
+    return sorted(set(conditions))
+
+
+def _line_conditions(line: str) -> list[str]:
+    return [line] if _looks_conditional(line) else []
+
+
+def _looks_conditional(text: str) -> bool:
+    lowered = text.lower()
+    return any(token in lowered for token in ("feature", "flag", "toggle", "enabled", "disabled", "experiment", "beta", "rollout", "paused"))
+
+
+def _js_handler_name(line: str) -> str | None:
+    if "function " in line:
+        return line.split("function ", 1)[1].split("(", 1)[0].strip() or None
+    if "=>" in line:
+        return "arrow_handler"
+    return None
+
+
+def _hint_from_result(result: object, analyzer: str, rules: dict[str, dict[str, object]]) -> StaticHint | None:
+    if not isinstance(result, dict):
+        return None
+    rule_id = str(result.get("ruleId", "unknown"))
+    location = _primary_location(result)
+    if location is None:
+        return None
+    artifact = location.get("physicalLocation", {}).get("artifactLocation", {})
+    region = location.get("physicalLocation", {}).get("region", {})
+    uri = artifact.get("uri") if isinstance(artifact, dict) else None
+    if not isinstance(uri, str) or not uri:
+        return None
+    rule = rules.get(rule_id, {})
+    message = result.get("message", {})
+    message_text = message.get("text") if isinstance(message, dict) else None
+    return StaticHint(
+        analyzer=analyzer,
+        rule_id=rule_id,
+        title=str(rule.get("name") or rule.get("shortDescription") or rule_id),
+        severity=_severity(result, rule),
+        path=uri.lstrip("./"),
+        start_line=region.get("startLine") if isinstance(region.get("startLine"), int) else None,
+        message=message_text if isinstance(message_text, str) else rule_id,
+        fingerprint=_fingerprint(result),
+        provenance=f"sarif:{analyzer}:{rule_id}",
+    )
+
+
+def _primary_location(result: dict[str, object]) -> dict[str, Any] | None:
+    locations = result.get("locations")
+    if not isinstance(locations, list) or not locations or not isinstance(locations[0], dict):
+        return None
+    return locations[0]
+
+
+def _rules_by_id(rules: object) -> dict[str, dict[str, object]]:
+    by_id: dict[str, dict[str, object]] = {}
+    for rule in _items(rules):
+        rule_id = rule.get("id") if isinstance(rule, dict) else None
+        if isinstance(rule_id, str):
+            by_id[rule_id] = rule
+    return by_id
+
+
+def _severity(result: dict[str, object], rule: dict[str, object]) -> str:
+    level = result.get("level")
+    if level == "error":
+        return "high"
+    if level == "warning":
+        return "medium"
+    properties = rule.get("properties")
+    if isinstance(properties, dict) and "security-severity" in properties:
+        try:
+            return "high" if float(properties["security-severity"]) >= 7.0 else "medium"
+        except (TypeError, ValueError):
+            return "medium"
+    return "low"
+
+
+def _fingerprint(result: dict[str, object]) -> str | None:
+    for key in ("partialFingerprints", "fingerprints"):
+        value = result.get(key)
+        if isinstance(value, dict) and value:
+            first = next(iter(value.values()))
+            return str(first)
+    return None
+
+
+def _normalise_analyzer(name: str) -> str:
+    lowered = name.lower()
+    if "semgrep" in lowered:
+        return "semgrep"
+    if "codeql" in lowered:
+        return "codeql"
+    return lowered.replace(" ", "_") or "sarif"
+
+
+def _items(value: object) -> list[object]:
+    return value if isinstance(value, list) else []
diff --git a/src/openultrasast/preprocess.py b/src/openultrasast/preprocess.py
index 765933e..cf433d2 100644
--- a/src/openultrasast/preprocess.py
+++ b/src/openultrasast/preprocess.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 import fnmatch
 import json
 import subprocess
-from dataclasses import asdict, dataclass
+from dataclasses import asdict, dataclass, field
 from pathlib import Path
 
 
@@ -29,6 +29,7 @@ LANGUAGE_BY_EXTENSION = {
     ".php": "php",
     ".cs": "csharp",
     ".swift": "swift",
+    ".sol": "solidity",
 }
 
 MEMORY_UNSAFE_LANGUAGES = {"c", "cpp"}
@@ -50,12 +51,22 @@ class FileTarget:
     loc: int
     tags: list[str]
     has_fuzz_entry_point: bool
+    static_hints: list[dict[str, object]] = field(default_factory=list)
+    reachability_hints: list[dict[str, object]] = field(default_factory=list)
 
 
-def preprocess_repository(root: Path, output_path: Path | None = None) -> tuple[RepoSnapshot, list[FileTarget]]:
+def preprocess_repository(
+    root: Path,
+    output_path: Path | None = None,
+    static_hints: list[object] | None = None,
+) -> tuple[RepoSnapshot, list[FileTarget]]:
     resolved = root.resolve()
     files = enumerate_source_files(resolved)
     targets = [build_file_target(resolved, path) for path in files]
+    if static_hints:
+        from .mapping import attach_static_hints
+
+        targets = attach_static_hints(targets, static_hints)  # type: ignore[arg-type]
     languages: dict[str, int] = {}
     for target in targets:
         languages[target.language] = languages.get(target.language, 0) + 1
@@ -68,13 +79,17 @@ def preprocess_repository(root: Path, output_path: Path | None = None) -> tuple[
     )
 
     if output_path is not None:
-        output_path.parent.mkdir(parents=True, exist_ok=True)
-        payload = {"snapshot": asdict(snapshot), "file_targets": [asdict(target) for target in targets]}
-        output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+        write_preprocess_artifact(snapshot, targets, output_path)
 
     return snapshot, targets
 
 
+def write_preprocess_artifact(snapshot: RepoSnapshot, targets: list[FileTarget], output_path: Path) -> None:
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+    payload = {"snapshot": asdict(snapshot), "file_targets": [asdict(target) for target in targets]}
+    output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
+
+
 def enumerate_source_files(root: Path) -> list[Path]:
     patterns = _load_ignore_patterns(root)
     paths: list[Path] = []
@@ -97,6 +112,8 @@ def build_file_target(root: Path, path: Path) -> FileTarget:
         loc=count_loc(text),
         tags=tags,
         has_fuzz_entry_point="LLVMFuzzerTestOneInput" in text,
+        static_hints=[],
+        reachability_hints=[],
     )
 
 
diff --git a/src/openultrasast/rank.py b/src/openultrasast/rank.py
index 9ebd933..298bc10 100644
--- a/src/openultrasast/rank.py
+++ b/src/openultrasast/rank.py
@@ -105,6 +105,10 @@ def heuristic_rank(target: FileTarget) -> RankingScore:
     if target.loc >= 500:
         influence = max(influence, 3)
         boosts.append("large_file")
+    reachability_boost = _reachability_from_hints(target.reachability_hints)
+    if reachability_boost:
+        reachability = max(reachability, reachability_boost)
+        boosts.append("entry_point_reachable")
 
     return RankingScore(
         path=target.path,
@@ -212,4 +216,23 @@ def _target_prompt_payload(target: FileTarget) -> dict[str, object]:
         "loc": target.loc,
         "tags": target.tags,
         "has_fuzz_entry_point": target.has_fuzz_entry_point,
+        "reachability_hints": target.reachability_hints,
     }
+
+
+def _reachability_from_hints(hints: list[dict[str, object]]) -> int:
+    score = 0
+    for hint in hints:
+        access_level = hint.get("access_level")
+        kind = hint.get("kind")
+        if access_level == "public":
+            score = max(score, 5 if kind in {"route", "parser", "fuzz"} else 4)
+        elif access_level in {"authenticated", "contract-only/callback"}:
+            score = max(score, 4)
+        elif access_level == "role-restricted":
+            score = max(score, 3)
+        elif access_level == "review-required":
+            score = max(score, 2)
+        elif access_level == "local-only":
+            score = max(score, 2)
+    return score
diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py
index 41b5909..412467d 100644
--- a/src/openultrasast/reports.py
+++ b/src/openultrasast/reports.py
@@ -18,9 +18,12 @@ def write_markdown_report(findings: list[StaticFinding], path: Path) -> None:
                 f"- ID: `{finding.finding_id}`",
                 f"- Path: `{finding.path}`",
                 f"- Line: `{finding.line}`",
+                f"- Function: `{finding.function_name or 'unknown'}`",
                 f"- Severity: `{finding.severity}`",
                 f"- Confidence: `{finding.confidence}`",
                 f"- Evidence level: `{finding.evidence_level}`",
+                f"- Reachability: `{finding.reachability_status}`",
+                f"- Conditions: `{', '.join(finding.reachability_conditions) or 'none'}`",
                 f"- Ranking priority: `{finding.ranking_priority}`",
                 f"- Tags: `{', '.join(finding.tags)}`",
                 "",
diff --git a/tests/test_calibration.py b/tests/test_calibration.py
index aca3065..a931155 100644
--- a/tests/test_calibration.py
+++ b/tests/test_calibration.py
@@ -25,6 +25,10 @@ def _finding(path: str = "auth/login.py") -> StaticFinding:
         evidence_level="static_corroboration",
         rationale="pattern matched",
         line=1,
+        function_name=None,
+        reachability_status="unknown",
+        reachability_evidence=[],
+        reachability_conditions=[],
         tags=["syscall_entry"],
         ranking_priority=4.0,
     )
diff --git a/tests/test_findings.py b/tests/test_findings.py
index 522960b..cfc3953 100644
--- a/tests/test_findings.py
+++ b/tests/test_findings.py
@@ -2,6 +2,7 @@ import json
 from pathlib import Path
 
 from openultrasast.findings import build_quick_hunter_prompt, quick_scan_findings, write_findings
+from openultrasast.mapping import analyze_entry_points, attach_reachability_hints
 from openultrasast.preprocess import preprocess_repository
 from openultrasast.rank import rank_targets
 from openultrasast.reports import write_markdown_report
@@ -38,6 +39,37 @@ def test_findings_and_report_artifacts(tmp_path: Path) -> None:
     assert "Subprocess shell execution" in report_path.read_text()
 
 
+def test_findings_inherit_only_matching_function_level_reachability(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "app.py").write_text(
+        "@app.route('/admin')\n"
+        "@login_required\n"
+        "def admin_upload():\n"
+        "    if feature_flags.enabled('dangerous_upload'):\n"
+        "        return eval(request.data)\n"
+        "    return 'disabled'\n"
+        "\n"
+        "def helper(user_input):\n"
+        "    return exec(user_input)\n"
+    )
+    _, targets = preprocess_repository(repo)
+    targets = attach_reachability_hints(targets, analyze_entry_points(repo, targets))
+
+    findings = quick_scan_findings(repo, targets, rank_targets(targets))
+
+    reachable = next(finding for finding in findings if finding.line == 5)
+    unreachable = next(finding for finding in findings if finding.line == 9)
+
+    assert reachable.function_name == "admin_upload"
+    assert reachable.reachability_status == "reachable"
+    assert reachable.reachability_evidence[0]["access_level"] == "authenticated"
+    assert reachable.reachability_conditions == ["feature_flags.enabled('dangerous_upload')"]
+    assert unreachable.function_name is None
+    assert unreachable.reachability_status == "unknown"
+    assert unreachable.ranking_priority <= 1.5
+
+
 def test_quick_hunter_prompt_is_bounded() -> None:
     repo = Path("/repo/example.py")
     target = next(iter(preprocess_repository(repo.parent)[1]), None) if repo.parent.exists() else None
diff --git a/tests/test_harness.py b/tests/test_harness.py
new file mode 100644
index 0000000..5ae8fc2
--- /dev/null
+++ b/tests/test_harness.py
@@ -0,0 +1,104 @@
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from openultrasast.config import load_config
+from openultrasast.harness import (
+    HarnessContractError,
+    HarnessEvent,
+    HarnessRuntime,
+    HarnessTraceWriter,
+    ProcessorSpec,
+    write_harness_config,
+)
+
+
+@dataclass
+class RecordingProcessor:
+    spec: ProcessorSpec
+
+    def handle(self, event: HarnessEvent, state: dict[str, Any]) -> dict[str, Any]:
+        return {"seen_stage": event.stage}
+
+
+@dataclass
+class BadWriteProcessor:
+    spec: ProcessorSpec
+
+    def handle(self, event: HarnessEvent, state: dict[str, Any]) -> dict[str, Any]:
+        return {"undeclared": event.event_type}
+
+
+def test_runtime_emits_stage_events_and_processor_state(tmp_path: Path) -> None:
+    trace_path = tmp_path / "events.jsonl"
+    processor = RecordingProcessor(
+        ProcessorSpec(name="recorder", version="1", handles=("stage_start",), writes=("seen_stage",))
+    )
+    runtime = HarnessRuntime(
+        scan_id="scan-1",
+        config=load_config(None),
+        trace_writer=HarnessTraceWriter(trace_path),
+        processors=[processor],
+    )
+
+    result = runtime.run_stage("preprocess", lambda: "ok")
+
+    events = [json.loads(line) for line in trace_path.read_text().splitlines()]
+    assert result == "ok"
+    assert runtime.state["seen_stage"] == "preprocess"
+    assert [event["event_type"] for event in events] == ["stage_start", "stage_start", "stage_end"]
+    assert events[1]["payload"] == {"changed_state": ["seen_stage"], "processor": "recorder"}
+
+
+def test_strict_contract_mode_rejects_undeclared_writes(tmp_path: Path) -> None:
+    processor = BadWriteProcessor(ProcessorSpec(name="bad", version="1", handles=("stage_start",), writes=()))
+    runtime = HarnessRuntime(
+        scan_id="scan-1",
+        config=load_config(None),
+        trace_writer=HarnessTraceWriter(tmp_path / "events.jsonl"),
+        processors=[processor],
+    )
+
+    with pytest.raises(HarnessContractError, match="undeclared writes"):
+        runtime.emit("stage_start", "rank", {})
+
+
+def test_warn_contract_mode_records_degradation(tmp_path: Path) -> None:
+    processor = BadWriteProcessor(ProcessorSpec(name="bad", version="1", handles=("stage_start",), writes=()))
+    runtime = HarnessRuntime(
+        scan_id="scan-1",
+        config=load_config(None),
+        trace_writer=HarnessTraceWriter(tmp_path / "events.jsonl"),
+        processors=[processor],
+        contract_mode="warn",
+    )
+
+    runtime.emit("stage_start", "rank", {})
+
+    assert "undeclared writes" in runtime.state["degradations"][0]
+    assert '"status": "degraded"' in (tmp_path / "events.jsonl").read_text()
+
+
+def test_harness_config_serializes_models_processors_and_prompt_hashes(tmp_path: Path) -> None:
+    config_path = tmp_path / "openultrasast.toml"
+    config_path.write_text("[models]\nranker = \"openrouter/ranker\"\n")
+    processor = RecordingProcessor(
+        ProcessorSpec(name="recorder", version="2026-01-01", handles=("stage_start",), writes=("seen_stage",))
+    )
+    output = tmp_path / "harness.json"
+
+    write_harness_config(
+        config=load_config(config_path),
+        processors=[processor],
+        contract_mode="strict",
+        path=output,
+        prompts={"ranker": "score files"},
+    )
+
+    payload = json.loads(output.read_text())
+    assert payload["model_roles"]["ranker"] == "openrouter/ranker"
+    assert payload["processor_versions"] == {"recorder": "2026-01-01"}
+    assert len(payload["prompt_hashes"]["ranker"]) == 64
diff --git a/tests/test_mapping.py b/tests/test_mapping.py
new file mode 100644
index 0000000..10857b0
--- /dev/null
+++ b/tests/test_mapping.py
@@ -0,0 +1,215 @@
+import json
+from pathlib import Path
+
+from openultrasast.mapping import (
+    analyze_entry_points,
+    attach_reachability_hints,
+    attach_static_hints,
+    codeql_mapping_tasks,
+    differential_mapping_record,
+    ingest_sarif,
+    semgrep_mapping_tasks,
+    sharp_edge_record,
+    verifier_evidence_candidates,
+    write_entry_points,
+    write_static_hints,
+)
+from openultrasast.preprocess import preprocess_repository
+
+
+def test_ingest_sarif_normalizes_semgrep_hints(tmp_path: Path) -> None:
+    sarif = tmp_path / "semgrep.sarif"
+    sarif.write_text(json.dumps(_sarif_payload("Semgrep", "python.lang.security.audit.eval", "app.py")))
+
+    hints = ingest_sarif(sarif)
+
+    assert len(hints) == 1
+    assert hints[0].analyzer == "semgrep"
+    assert hints[0].rule_id == "python.lang.security.audit.eval"
+    assert hints[0].path == "app.py"
+    assert hints[0].start_line == 7
+    assert hints[0].severity == "high"
+    assert hints[0].provenance == "sarif:semgrep:python.lang.security.audit.eval"
+
+
+def test_static_hints_attach_to_file_targets_and_evidence_candidates(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "app.py").write_text("eval(user_input)\n")
+    hints = ingest_sarif(_write_sarif(tmp_path, "CodeQL", "py/code-injection", "app.py"))
+    _, targets = preprocess_repository(repo, static_hints=hints)
+
+    enriched = attach_static_hints(targets, hints)
+    evidence = verifier_evidence_candidates(hints, path="app.py")
+
+    assert enriched[0].static_hints[0]["analyzer"] == "codeql"
+    assert evidence == [
+        {
+            "source": "codeql",
+            "rule_id": "py/code-injection",
+            "path": "app.py",
+            "line": 7,
+            "severity": "high",
+            "message": "Potential injection",
+            "provenance": "sarif:codeql:py/code-injection",
+        }
+    ]
+
+
+def test_mapping_task_records_cover_disciplines(tmp_path: Path) -> None:
+    semgrep_hints = ingest_sarif(_write_sarif(tmp_path, "Semgrep", "sg.rule", "parser.c"))
+    codeql_hints = ingest_sarif(_write_sarif(tmp_path, "CodeQL", "cpp/overflow", "parser.c"))
+
+    assert semgrep_mapping_tasks(semgrep_hints)[0].task == "pattern_variant_review"
+    assert codeql_mapping_tasks(codeql_hints)[0].task == "source_sink_sanitizer_path_review"
+    assert differential_mapping_record(
+        "parser.c",
+        change_kind="modified_function",
+        trust_boundary="network_input",
+        rationale="Parser changed behind a socket boundary.",
+    ).trust_boundary == "network_input"
+    assert sharp_edge_record(
+        "config.py",
+        category="insecure_default",
+        surface="debug mode",
+        rationale="Debug default changes production risk.",
+    ).category == "insecure_default"
+
+
+def test_write_static_hints_artifact(tmp_path: Path) -> None:
+    hints = ingest_sarif(_write_sarif(tmp_path, "Semgrep", "sg.rule", "app.py"))
+    output = tmp_path / "mapping" / "static_hints.json"
+
+    write_static_hints(hints, output)
+
+    assert json.loads(output.read_text())["static_hints"][0]["analyzer"] == "semgrep"
+
+
+def test_entry_point_mapping_classifies_routes_and_attaches_reachability(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "app.py").write_text("@app.route('/upload')\ndef upload():\n    return parse(request.data)\n")
+    _, targets = preprocess_repository(repo)
+
+    entry_points = analyze_entry_points(repo, targets)
+    enriched = attach_reachability_hints(targets, entry_points)
+
+    assert entry_points[0].kind == "route"
+    assert entry_points[0].access_level == "public"
+    assert enriched[0].reachability_hints[0]["trust_boundary"] == "http_request"
+
+
+def test_entry_point_mapping_distinguishes_authenticated_routes_and_conditions(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "app.py").write_text(
+        "@app.route('/admin')\n"
+        "@login_required\n"
+        "def admin_upload():\n"
+        "    if feature_flags.enabled('dangerous_upload'):\n"
+        "        return eval(request.data)\n"
+        "    return 'disabled'\n"
+    )
+    _, targets = preprocess_repository(repo)
+
+    entry_points = analyze_entry_points(repo, targets)
+
+    route = next(entry_point for entry_point in entry_points if entry_point.kind == "route")
+    assert route.function_name == "admin_upload"
+    assert route.line == 3
+    assert route.end_line == 6
+    assert route.access_level == "authenticated"
+    assert route.access_evidence == ["login_required"]
+    assert route.conditions == ["feature_flags.enabled('dangerous_upload')"]
+
+
+def test_entry_point_mapping_covers_solidity_state_changing_access(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "Vault.sol").write_text(
+        "contract Vault {\n"
+        "  function withdraw(uint256 amount) external onlyOwner { }\n"
+        "  function balance() external view returns (uint256) { return 0; }\n"
+        "}\n"
+    )
+    _, targets = preprocess_repository(repo)
+
+    entry_points = analyze_entry_points(repo, targets)
+
+    assert len(entry_points) == 1
+    assert entry_points[0].name == "withdraw"
+    assert entry_points[0].access_level == "role-restricted"
+
+
+def test_entry_point_mapping_records_solidity_feature_conditions(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "Vault.sol").write_text(
+        "contract Vault {\n"
+        "  function withdraw(uint256 amount) external onlyOwner whenNotPaused { }\n"
+        "}\n"
+    )
+    _, targets = preprocess_repository(repo)
+
+    entry_points = analyze_entry_points(repo, targets)
+
+    assert entry_points[0].access_level == "role-restricted"
+    assert entry_points[0].access_evidence == ["onlyOwner"]
+    assert entry_points[0].conditions == ["whenNotPaused"]
+
+
+def test_write_entry_points_artifact(tmp_path: Path) -> None:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "cli.py").write_text("if __name__ == '__main__':\n    main()\n")
+    _, targets = preprocess_repository(repo)
+    output = tmp_path / "mapping" / "entry_points.json"
+
+    write_entry_points(analyze_entry_points(repo, targets), output)
+
+    payload = json.loads(output.read_text())
+    assert payload["entry_points"][0]["kind"] == "cli"
+
+
+def _write_sarif(tmp_path: Path, analyzer: str, rule_id: str, result_path: str) -> Path:
+    path = tmp_path / f"{analyzer.lower()}.sarif"
+    path.write_text(json.dumps(_sarif_payload(analyzer, rule_id, result_path)))
+    return path
+
+
+def _sarif_payload(analyzer: str, rule_id: str, result_path: str) -> dict[str, object]:
+    return {
+        "version": "2.1.0",
+        "runs": [
+            {
+                "tool": {
+                    "driver": {
+                        "name": analyzer,
+                        "rules": [
+                            {
+                                "id": rule_id,
+                                "name": "Injection rule",
+                                "properties": {"security-severity": "8.1"},
+                            }
+                        ],
+                    }
+                },
+                "results": [
+                    {
+                        "ruleId": rule_id,
+                        "level": "error",
+                        "message": {"text": "Potential injection"},
+                        "partialFingerprints": {"primaryLocationLineHash": "abc123"},
+                        "locations": [
+                            {
+                                "physicalLocation": {
+                                    "artifactLocation": {"uri": result_path},
+                                    "region": {"startLine": 7},
+                                }
+                            }
+                        ],
+                    }
+                ],
+            }
+        ],
+    }
diff --git a/tests/test_rank.py b/tests/test_rank.py
index c56791e..5c03657 100644
--- a/tests/test_rank.py
+++ b/tests/test_rank.py
@@ -110,3 +110,26 @@ def test_write_rankings_emits_artifact(tmp_path: Path) -> None:
     payload = json.loads(output.read_text())
     assert payload["rankings"][0]["path"] == "auth.py"
     assert payload["rankings"][0]["priority"] >= 1
+
+
+def test_reachability_hints_raise_ranking_priority() -> None:
+    target = FileTarget(
+        path="routes.py",
+        absolute_path="/repo/routes.py",
+        language="python",
+        loc=5,
+        tags=[],
+        has_fuzz_entry_point=False,
+        reachability_hints=[
+            {
+                "kind": "route",
+                "access_level": "public",
+                "trust_boundary": "http_request",
+            }
+        ],
+    )
+
+    ranking = rank_targets([target])[0]
+
+    assert ranking.reachability == 5
+    assert "entry_point_reachable" in ranking.static_boosts