phases

func BindArchitectureMap ¶

func BindArchitectureMap(payload map[string]any) (schemas.ArchitectureMap, error)

BindArchitectureMap is `ArchitectureMap(**architecture)` (reasoners/recon.py:43 and :52). ArchitectureMap has no required field of its OWN, but each of its five lists holds a model that does, so an `{"modules": [{"name": "x"}]}` payload raises in Python and must raise here.

func BindConfigReport ¶

func BindConfigReport(payload map[string]any) (schemas.ConfigReport, error)

BindConfigReport is `ConfigReport.model_validate(payload)` (reasoners/phases.py:177). Same shape as BindDependencyReport: no required field of its own, but `SecretFinding` requires secret_type/file_path/line/ match/confidence and `MisconfigFinding` requires category/file_path/risk.

func BindDataFlowMap ¶

func BindDataFlowMap(payload map[string]any) (schemas.DataFlowMap, error)

BindDataFlowMap is `DataFlowMap.model_validate(payload)` (reasoners/phases.py:186). `DataFlow` requires source/sink/sanitized, `Sink` requires sink_type/file_path/line and `SanitizationPoint` requires file_path/line/sanitization_type, so a flow missing `sanitized` raises in Python (VERIFIED) where afx.Bind alone would default-seed it to false.

func BindDataFlowTrace ¶

func BindDataFlowTrace(payload map[string]any) (schemas.DataFlowTrace, error)

BindDataFlowTrace is `DataFlowTrace(**data_flow)` (reasoners/prove.py:77).

`source`, `sink` and `steps` are required but carry `mode="before"` validators, which pydantic runs AHEAD of the required/type check — so an explicit null on any of them is legal and coerces to "unknown"/[] (the schemas package reproduces that coercion). Only a MISSING key raises.

func BindDependencyReport ¶

func BindDependencyReport(payload map[string]any) (schemas.DependencyReport, error)

BindDependencyReport is `DependencyReport.model_validate(payload)` (reasoners/phases.py:171). DependencyReport has no required field of its OWN, but each of its three lists holds a model that does — `Dependency` requires name/version/ecosystem/direct — so an `{"sbom": [{"name": "a"}]}` payload raises in Python (VERIFIED: 3 errors) and must raise here.

func BindExploitHypothesis ¶

func BindExploitHypothesis(payload map[string]any) (schemas.ExploitHypothesis, error)

BindExploitHypothesis is `ExploitHypothesis(**exploit)` (reasoners/prove.py:91). Both required fields are before-validated, so no field of this model rejects an explicit null.

func BindFindingForVerifier ¶

func BindFindingForVerifier(payload map[string]any) (schemas.FindingForVerifier, error)

BindFindingForVerifier is `FindingForVerifier.model_validate(finding)` — the fallback branch of `_coerce_verifier_finding` (reasoners/prove.py:28).

func BindHuntResult ¶

func BindHuntResult(payload map[string]any) (schemas.HuntResult, error)

BindHuntResult is `HuntResult.model_validate(payload)`. HuntResult itself has no required field, but every element of its `findings` and `chains` lists is a model that does — and both carry enum-typed fields, which Go's plain-string enums would otherwise let through. An out-of-vocabulary finding_type/estimated_severity/confidence is a ValidationError in Python, and each call site turns that into a different observable outcome: an empty batch plus a "Hunt strategy failed: X: ..." note in hunt_phase's per-hunter try (reasoners/phases.py:303), a failed phase in prove_phase (:425), and an unexpected sixth key in SecurityAuditResult.by_severity downstream.

All seven of its fields also reject an explicit null: `{"findings": null}` raises in Python, where afx.Bind alone would accept it AND wipe the seeded `[]`, emitting `"findings": null` from the phase.

func BindRawFinding ¶

func BindRawFinding(payload map[string]any) (schemas.RawFinding, error)

BindRawFinding is `RawFinding(**finding)` / `RawFinding.model_validate(finding)`.

Exported because internal/reasoners binds the SAME model at its own `.call` boundaries (run_verifier_agent, run_remediation_agent, …). One spec, one check: a second transcription of RawFinding's required fields could drift from this one and silently change which branch `_coerce_verifier_finding` takes.

func BindReconResult ¶

func BindReconResult(payload map[string]any) (schemas.ReconResult, error)

BindReconResult is `ReconResult(**recon_context)`.

All five nested models are validated, not just `security_context`: Python raises on `{"architecture": {"modules": [{"name": "x"}]}, ...}` because `Module` requires path and language too, and the same holds for `data_flows.flows[]`, `dependencies.sbom[]` and `config.secrets[]`. hunt_phase (reasoners/phases.py:253) and run_deduplicator (hunt.py:287) both take that raise as a hard failure.

func BindRemediationSuggestion ¶

func BindRemediationSuggestion(payload map[string]any) (schemas.RemediationSuggestion, error)

BindRemediationSuggestion is `RemediationSuggestion.model_validate(payload)`.

remediation_phase wraps the call in `except Exception: pass`, so a failure here means the finding simply keeps `remediation = None` and does not count toward the "N/M generated" tally.

func BindSanitizationResult ¶

func BindSanitizationResult(payload map[string]any) (schemas.SanitizationResult, error)

BindSanitizationResult is `SanitizationResult(**sanitization)` (reasoners/prove.py:84). `found` has no before-validator, so a null there raises just as a missing key does.

func BindSecurityContext ¶

func BindSecurityContext(payload map[string]any) (schemas.SecurityContext, error)

BindSecurityContext is `SecurityContext.model_validate(payload)`.

func BindVerifiedFinding ¶

func BindVerifiedFinding(payload map[string]any) (schemas.VerifiedFinding, error)

BindVerifiedFinding is `VerifiedFinding.model_validate(payload)`: the required-field, null, enum and NESTED-MODEL checks pydantic performs, then afx.Bind.

The nested layer matters. A payload whose `proof` is `{}` or whose `cvss_v4` carries only a `vector` binds cleanly through afx.Bind alone — producing a Proof with `evidence_level: 0`, which is not a member of the EvidenceLevel IntEnum — while `VerifiedFinding.model_validate` on the same dict raises. The difference is observable: reasoners/phases.py:496 catches that raise to demote the finding (drop_reason "schema_parse_failure", verdict inconclusive, a drop_summary entry and a "Demoted finding ..." note), so without the nested checks the Go node would ship the prover's reported verdict where Python demotes it.

func CWEExpansionPrompt ¶

func CWEExpansionPrompt(reconSummary string, strategies []string) string

CWEExpansionPrompt builds the exact user prompt `expand_cwes_for_hunt` sends to `.ai(schema=CWEExpansion)` (reasoners/phases.py:117).

It is a separate function only so the golden test can compare the bytes without scripting an AI response; ExpandCWEsForHunt is the single caller. `", ".join(strategies)` renders an empty list as the empty string, so an empty strategy list yields the line "Active strategies: " with a trailing space.

func DefaultStrategies ¶

func DefaultStrategies(recon schemas.ReconResult, depth string) []schemas.HuntStrategy

DefaultStrategies ports `_default_strategies(recon, depth)` (reasoners/phases.py:214) — the hunt_phase variant, which is NOT the same list as AuditOrchestrator._default_strategies:

base: injection, dos, ssrf, auth, data_exposure, config_secrets
+ xss             when depth is standard or thorough      <-- phases only
+ crypto          when security_context.crypto_usage
+ supply_chain    when dependencies.direct_count > 0
+ api_security    when architecture.api_surface
+ business_logic  when depth is standard or thorough

Differences from the orchestrator's copy, both deliberate: this one adds XSS (the orchestrator's never does) and never adds PYTHON_SPECIFIC / JAVASCRIPT_SPECIFIC (the orchestrator's does at thorough depth).

The trailing de-duplication pass preserves FIRST-seen order, which is a no-op for the list above — no branch can add a strategy twice — but is reproduced because it is what Python guarantees.

func ExpandCWEsForHunt ¶

func ExpandCWEsForHunt(ctx context.Context, app appx.AIer, reconSummary string, strategies []string) []string

ExpandCWEsForHunt ports `expand_cwes_for_hunt` (reasoners/phases.py:109):

try:
    result = await router.ai(user=prompt, schema=CWEExpansion)
    ...
    return expansion.additional_cwes
except Exception:
    return []            # graceful fallback — AI expansion is additive

Python parity:

• EVERY failure is swallowed and yields an empty list. That is the whole point of the helper: a dead AI gate must not fail the hunt. Go therefore returns no error — the signature carries the same information the Python caller has.
• the two `isinstance` branches are Python duck-typing over an SDK that may hand back either the model or a raw dict; aix.Structured always parses into the typed value, so both collapse into one path.
• a successful call whose `additional_cwes` is absent yields `[]` (pydantic default_factory=list, reproduced by the schemas package's default seeding), NOT nil — the caller's `if additional_cwes:` guard treats them the same, but the returned slice is non-nil so a JSON round trip spells it "[]".

func HuntPhase ¶

func HuntPhase(
	ctx context.Context,
	app appx.App,
	nodeID string,
	repoPath string,
	reconContext map[string]any,
	depth string,
	aiGate StrategySelector,
	maxConcurrentHunters int,
	earlyStopFileThreshold int,
) (map[string]any, error)

HuntPhase ports the `hunt_phase` reasoner (reasoners/phases.py:243):

@router.reasoner()
async def hunt_phase(repo_path, recon_context, depth="standard", ai_gate=None,
                     max_concurrent_hunters=4, early_stop_file_threshold=30)

Sequence:

1. `ReconResult(**recon_context)` — pydantic ignores unknown keys and seeds defaults for missing ones, which afx.Bind reproduces through the schemas package's default-seeding UnmarshalJSON.
2. `_default_strategies(recon, depth)`; optionally narrowed by the AI gate.
3. `expand_cwes_for_hunt(recon_context_generic(recon), [s.value ...])` — an `.ai()` call, NOT a `.call`, so it adds no DAG node. Note the summary is `recon_context_generic`, a DIFFERENT function from the `_recon_summary_string` fed to the AI gate one step earlier.
4. one producer per strategy under `Semaphore(max(1, min(max_concurrent_hunters, N)))`, each `.call`ing run_<strategy>_hunter and publishing its findings to a queue; one consumer draining exactly N batches and de-duplicating by fingerprint as they arrive.
5. `run_deduplicator` — ONLY when at least one fingerprint-unique finding survived. An empty hunt makes no call at all.
6. `strategies_run` and `total_raw` are overwritten AFTER the dedup call, so the deduplicator's own values for those two fields are discarded.

Python parity notes:

• the AI-gate result is mapped back through `{s.value: s for s in default_candidates}` and unknown names are DROPPED; a selection that maps to nothing falls back to the full default list with a note. A name repeated in the selection yields a repeated strategy — and therefore a repeated hunter call — which is reproduced.
• the CWE expansion result is only ever used for a note. Nothing downstream consumes `additional_cwes`; the hunters keep their hardcoded baselines.
• a failing hunter is noted and contributes an EMPTY batch, so the consumer still terminates. The error text is Go's, not Python's `str(exc)`.
• `early_stop_file_threshold` becomes the `max_files_without_signal` kwarg — unlike agents/hunt's identically named parameter, this one is observable.
• `recon_context` is forwarded to run_deduplicator VERBATIM (the caller's full dump), while each hunter receives the STRATEGY-PRUNED projection.

Concurrency parity: Python's `asyncio.Queue` is unbounded and its consumer is a peer task. The Go channel is buffered to the strategy count so a producer never blocks either, and the consumer goroutine reads exactly that many batches — the same termination condition as `while completed < len(strategies)`.

func NodeID ¶

func NodeID() string

NodeID returns DefaultNodeID. It stays a function so the call sites that threaded the environment read through the phase functions keep their shape.

func ProvePhase ¶

func ProvePhase(
	ctx context.Context,
	app appx.App,
	nodeID string,
	repoPath string,
	huntResult map[string]any,
	depth string,
	maxProvers *int,
	maxConcurrentProvers int,
) (map[string]any, error)

ProvePhase ports the `prove_phase` reasoner (reasoners/phases.py:416):

@router.reasoner()
async def prove_phase(repo_path, hunt_result, depth="standard",
                      max_provers=None, max_concurrent_provers=3)

One `.call` to run_verifier per selected finding, fanned out under `Semaphore(max(1, min(max_concurrent_provers, len(selected))) if selected else 1)`, gathered with return_exceptions=True, then a SEQUENTIAL pass over the results in `selected` order that classifies each into one of four outcomes:

.call raised                       -> drop "verifier_error"
_unwrap/_as_dict raised            -> drop "schema_parse_failure"
payload["verdict"] == "unverified" -> drop "verdict_unverified"
model_validate raised              -> drop "schema_parse_failure"
                                      (with original_verdict when present)

Every drop keeps the finding in the report as a demoted INCONCLUSIVE VerifiedFinding built by agents/prove.Fallback, so nothing is silently lost.

Python parity notes:

• the drop pass is sequential and ordered, so the drop_summary entries and their notes appear in `selected` order regardless of completion order.
• the "unverified" test is `isinstance(v, str) and v.lower() == "unverified"`: case-insensitive, and a non-string verdict skips straight to model_validate (which then fails, taking the fourth branch).
• `original_verdict` is the RAW value on the unverified branch and `str(verdict_value)` on the model_validate branch — the same string for a JSON string, which is the only shape either branch sees in practice.
• the result's `verified` list is `model_dump(exclude_none=True)`, so absent optionals (cvss_v4, proof, remediation, drop_reason, …) are OMITTED rather than null.
• `not_verified` is `max(0, total - len(selected))`, which is the count the orchestrator copies into its own findings_not_verified.

func ReconPhase ¶

func ReconPhase(ctx context.Context, app appx.App, nodeID, repoPath, depth string) (map[string]any, error)

ReconPhase ports the `recon_phase` reasoner (reasoners/phases.py:153):

@router.reasoner()
async def recon_phase(repo_path: str, depth: str = "standard") -> dict[str, Any]:

The DAG it draws — three unconditional children, then two more unless the depth normalizes to quick:

arch_raw, deps_raw, config_raw = await asyncio.gather(
    router.call(f"{NODE_ID}.run_architecture_mapper", repo_path=repo_path),
    router.call(f"{NODE_ID}.run_dependency_auditor",  repo_path=repo_path),
    router.call(f"{NODE_ID}.run_config_scanner",      repo_path=repo_path),
)
if _normalize_depth(depth) == QUICK:
    data_flows = DataFlowMap(); security_context = SecurityContext(auth_model="unknown", auth_details="unknown")
else:
    flow_raw, sec_raw = await asyncio.gather(
        router.call(f"{NODE_ID}.run_data_flow_mapper",            repo_path=..., architecture=architecture.model_dump()),
        router.call(f"{NODE_ID}.run_security_context_profiler",   repo_path=..., architecture=architecture.model_dump()),
    )

Python parity notes:

• the QUICK placeholders are the SAME pair agents/recon._quick_defaults builds — `auth_model="unknown", auth_details="unknown"`, both literal "unknown". They are NOT the `{"auth_model": "unknown", "auth_details": ""}` seed that reasoners/recon.py's `_recon_model` uses; do not conflate them.
• `architecture.model_dump()` is computed TWICE in Python (once per call); afx.ToMap is called once here and the same map is handed to both calls. The map is never mutated, and the SDK marshals it per call, so the two request bodies are identical either way.
• recon_duration_seconds is NOT set: `ReconResult(...)` omits it, so the returned dict carries the pydantic default 0.0. app.py overwrites it on the caller side with its own stopwatch.
• _repo_metrics walks the repository on the phase's own process, not through a `.call` — no DAG node.

Concurrency parity: `asyncio.gather` with the default return_exceptions=False surfaces the first exception; errgroup.Group (deliberately WITHOUT WithContext, so a failing sibling is not cancelled — gather does not cancel either) returns the first error by completion time. The one difference is that Wait blocks until all goroutines finish while `await gather` resumes the caller immediately and leaves the rest detached; the value returned is the same, and the caller's next act on error is to propagate it.

func ReconSummaryString ¶

func ReconSummaryString(recon schemas.ReconResult) string

ReconSummaryString ports `_recon_summary_string` (reasoners/phases.py:59) — the natural-language recon digest hunt_phase feeds to the AI strategy gate:

"python/typescript (django), 5000 LOC, 42 files, jwt auth, 5 direct dependencies"

Every guard is Python TRUTHINESS, which is where the subtleties live:

• `if recon.languages` / `if recon.frameworks` — an empty list contributes nothing, and the frameworks parenthetical only appears when BOTH are non-empty. Neither list is filtered, so a blank entry survives into the "/"-join.
• `if recon.security_context.auth_model` — an EMPTY auth model is skipped entirely; it is not rendered as " auth".
• the three counter guards are `> 0` comparisons, so a negative count (impossible from pydantic, but not from a hand-built struct) is skipped rather than rendered.
• the fallback for an all-empty recon is the literal "Unknown application".

The integers are Python `str(int)`, i.e. plain decimal — no thousands separators — which strconv.Itoa reproduces.

func RemediationPhase ¶

func RemediationPhase(
	ctx context.Context,
	app appx.App,
	nodeID string,
	repoPath string,
	verifiedFindings []map[string]any,
	maxConcurrentRemediations int,
) (map[string]any, error)

RemediationPhase ports the `remediation_phase` reasoner (reasoners/phases.py:529):

@router.reasoner()
async def remediation_phase(repo_path, verified_findings,
                            max_concurrent_remediations=3)

findings = [VerifiedFinding.model_validate(v) for v in verified_findings]
needs = [(i, f) for i, f in enumerate(findings)
         if f.verdict in {CONFIRMED, LIKELY} and f.remediation is None]
if not needs: note("No findings need remediation"); return {"verified": [...]}
sem = Semaphore(max(1, min(max_concurrent_remediations, len(needs))))
results = await gather(*[_call_remediation(i, f) for i, f in needs])
for i, payload in results:
    if payload is not None:
        try: findings[i].remediation = RemediationSuggestion.model_validate(payload); generated += 1
        except Exception: pass

Python parity notes:

• the eligibility filter is `verdict in {CONFIRMED, LIKELY}` AND `remediation is None`. A finding that already carries a remediation is left untouched and does not count toward the "N/M generated" tally.
• `_call_remediation` swallows EVERY exception into `(idx, None)` — a failed `.call`, a failed _unwrap and a failed _as_dict are indistinguishable, and none of them emits a note.
• the finding handed to run_remediation is `finding.model_dump()`, the FULL dump (nulls included), whereas the phase's own return value is `model_dump(exclude_none=True)`.
• a payload that fails RemediationSuggestion validation is silently dropped (`except Exception: pass`), leaving the finding without a remediation.
• the early-return branch emits the "No findings need remediation" note with the DONE tags and skips the completion note entirely.

Concurrency parity: `asyncio.gather` over the eligible findings under a semaphore; results are written into pre-indexed slots so the apply pass runs in `needs` order regardless of completion order (Python's gather preserves argument order the same way).

func RunCWEExpansion ¶

func RunCWEExpansion(ctx context.Context, app appx.AIer, reconSummary string, strategies []string) map[string]any

RunCWEExpansion ports the `run_cwe_expansion` reasoner body (reasoners/phases.py:140):

additional = await expand_cwes_for_hunt(recon_summary, strategies)
return {"additional_cwes": additional}

Registered on the router but never `.call`ed by any phase — hunt_phase calls expand_cwes_for_hunt in process — so it contributes no DAG node. It returns no error for the same reason ExpandCWEsForHunt does not.