orch

package
v0.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 14, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Overview

Package orch ports src/sec_af/orchestrator.py — the AuditOrchestrator class, its budget/cost bookkeeping, its checkpoint format, and the output-generation step every entry point funnels through.

Two paths through this package

The class has TWO ways of driving an audit, and only one of them is reached by the live REST API:

  • Run (`AuditOrchestrator.run`) is the STREAMING, in-process pipeline: fast recon, then deep recon / hunt / prove concurrently, with hunt publishing findings to prove through a queue. Every sub-agent is a Go function call, so the control plane sees ONE execution. app.py does not use it — it issues four `.call`s into internal/phases instead — but RunFromCheckpoint shares its `_run_hunt` / `_run_prove` halves, so the whole thing is ported.
  • GenerateOutput is the shared tail: severity filtering, the CWE severity floor, exploitability scoring, compliance mapping, the counters, the SARIF/JSON/Markdown artifacts. Both paths end here, and app.py calls it directly after its four `.call`s.

What the node wiring must know

app.py constructs the orchestrator and then OVERWRITES two fields: 

orchestrator = AuditOrchestrator(app=app, input=audit_input)
repo_path = _resolve_repo(repo_url)
orchestrator.repo_path = Path(repo_path)
orchestrator.checkpoint_dir = orchestrator.repo_path / ".sec-af"

so New computes its own repo path from SEC_AF_REPO_PATH/cwd (Python parity — including running the PR-mode diff analysis against THAT path, before the override) and RepoPath/CheckpointDir are exported for the caller to replace. SetRepoPath does both assignments in one call.

codeaf's audit entry (internal/secaf/audit) does not take that road: it resolves the repository first and constructs with NewAt, so the diff analysis and AuditConfig.repo_path describe the audited repository. New and NewWithContext keep the Python order for the in-process Run path.

app.py likewise writes agent_invocations, findings_not_verified and prove_drop_summary from the phase results before calling GenerateOutput. FindingsNotVerified and ProveDropSummary are exported fields for the same reason; the invocation counter is mutex-guarded (see below), so it is written through SetAgentInvocations instead.

Concurrency

Python's orchestrator is single-threaded asyncio, so its counters need no locking. The Go port fans phases out across goroutines and the phase proxy is shared by all of them, so the cost/invocation bookkeeping is mutex-guarded. All four guarded values are unexported and reachable only through TotalCostUSD / CostBreakdown / AgentInvocations / SetAgentInvocations / BudgetExhausted, so the lock contract cannot be bypassed by a later caller that reads a partial result while a phase is still running. That is the only structural addition; every value it computes is the value Python computes.

Index

Constants

View Source 
const (
	PhaseRecon = "recon"
	PhaseHunt  = "hunt"
	PhaseProve = "prove"
)

The three phase names, spelled once so a typo cannot silently create a fourth cost bucket.

Variables

View Source 
var ErrDastVerifierArity = errors.New(
	"run_dast_verifier() missing 2 required positional arguments: 'exploit_payload' and 'depth'")

ErrDastVerifierArity is the TypeError orchestrator.py:341 raises.

PYTHON BUG, REPRODUCED. `_run_dast_verification` calls 

await run_dast_verifier(_PhaseHarnessProxy(self, "prove"), str(self.repo_path), finding)

with THREE positional arguments, while agents/prove/dast_verifier.py declares 

async def run_dast_verifier(app, repo_path, finding, exploit_payload, depth)

— five required parameters. The call raises before the agent runs, the surrounding `except Exception` catches it, and the finding is tagged "dast_error". (Two further mismatches sit behind it: `finding` is a VerifiedFinding where a RawFinding is expected, and the result attributes the loop body reads — exploit_attempted, exploit_succeeded, evidence, confidence, response_analysis — do not exist on DastVerificationResult, which carries payload_sent / response_summary / exploit_confirmed / safety_notes.)

View Source 
var PhaseOrder = [...]string{"recon", "hunt", "prove"}

PhaseOrder ports `AuditOrchestrator._PHASE_ORDER = ("recon", "hunt", "prove")` — the three cost buckets, and the key set of cost_breakdown.

Functions

func IsBudgetExhausted 

func IsBudgetExhausted(err error) bool

IsBudgetExhausted reports whether err is (or wraps) a BudgetExhaustedError, which is Go's `except BudgetExhausted`.

func MergeReconFindingsIntoHunt 

func MergeReconFindingsIntoHunt(hunt schemas.HuntResult, reconFindings []schemas.RawFinding) schemas.HuntResult

MergeReconFindingsIntoHunt ports the module-level `merge_recon_findings_into_hunt` (orchestrator.py:700) — how the findings RECON detected on its own (hardcoded secrets, misconfigs, weak TLS) join the hunters' output: 

if not recon_findings: return hunt
merged_findings = [*recon_findings, *hunt.findings]
strategies_run = list(hunt.strategies_run)
if "recon" not in strategies_run: strategies_run.insert(0, "recon")
return HuntResult(findings=merged_findings, chains=hunt.chains,
                  total_raw=hunt.total_raw + len(recon_findings),
                  deduplicated_count=len(merged_findings),
                  chain_count=hunt.chain_count, strategies_run=strategies_run,
                  hunt_duration_seconds=hunt.hunt_duration_seconds)

Behaviours tests/test_recon_findings.py::test_merge_recon_findings_prepends_and_updates_counts pins, all reproduced:

  • recon findings are PREPENDED, so findings[0] is a recon finding;
  • total_raw GROWS by the recon count (it is not recomputed);
  • deduplicated_count is REPLACED by the merged length — no dedup pass runs, so a recon finding that duplicates a hunter finding survives twice;
  • "recon" is inserted at the FRONT of strategies_run, and only when absent.

Python parity: an empty recon_findings list returns the hunt result UNCHANGED — the same value, not a copy — so a caller that mutates the result mutates the input. Go returns the same struct value, which copies the header but shares the backing arrays: identical aliasing.

func NewDropSummary 

func NewDropSummary() map[string]any

NewDropSummary builds the `{"demoted_total": 0, "by_reason": {}, "findings": []}` literal the orchestrator resets prove_drop_summary to.

func ReachabilitySummary 

func ReachabilitySummary(finding schemas.VerifiedFinding) string

ReachabilitySummary builds the exact prompt body `_assess_one` hands to AIGateWrapper.assess_reachability (orchestrator.py:580): 

summary = (f"Finding: {finding.title}\n"
           f"Description: {finding.description}\n"
           f"CWE: {finding.cwe_id}\n"
           f"File: {finding.location.file_path}:{finding.location.start_line}\n"
           f"Verdict: {finding.verdict.value}")

No trailing newline. `verdict.value` is the enum's string value, which the Go enum already is. It is a separate function so the golden test can compare the bytes without scripting a gate.

func ReadCheckpoint 

func ReadCheckpoint[T any](o *AuditOrchestrator, phase string, bind func(map[string]any) (T, error)) (T, error)

ReadCheckpoint ports `_read_checkpoint(phase, schema)`: 

payload = json.loads(path.read_text(encoding="utf-8"))
return schema(**payload.get("data", {}))

bind is the `schema(**data)` half: one of internal/phases' checked binders, so a hand-edited or stale checkpoint fails HERE — with a ValueError-class error the audit handler maps to HTTP 400 — instead of quietly binding to default-seeded values. Without it, a checkpoint whose findings carry an out-of-vocabulary severity resumed cleanly in Go and raised ValidationError in Python.

Python parity: a checkpoint whose "data" key is MISSING falls back to `{}`, which for a model with required fields raises and for one with all-defaults yields the defaults. Passing `{}` to the binder reproduces both.

func ReadCheckpointList 

func ReadCheckpointList[T any](o *AuditOrchestrator, phase string, bind func(map[string]any) (T, error)) ([]T, error)

ReadCheckpointList ports `_read_checkpoint_list(phase, schema)`: 

rows = payload.get("data", [])
return [schema(**row) for row in rows]

Same story as ReadCheckpoint: `schema(**row)` validates every element, so the binder runs per row and the first failure propagates.

func VerifiedFindingFallback 

func VerifiedFindingFallback(finding schemas.RawFinding) schemas.VerifiedFinding

VerifiedFindingFallback ports the module-level `_verified_finding_fallback` (orchestrator.py:672) — a RawFinding promoted to an INCONCLUSIVE VerifiedFinding when no proof is available. 

VerifiedFinding(
    id=finding.id, fingerprint=finding.fingerprint, title=..., description=...,
    finding_type=..., cwe_id=..., cwe_name=..., owasp_category=...,
    tags=[], verdict=Verdict.INCONCLUSIVE, evidence_level=EvidenceLevel.STATIC_MATCH,
    rationale="Automated proof unavailable; requires manual review.",
    severity=finding.estimated_severity, exploitability_score=0.0,
    location=Location(file_path=..., start_line=..., end_line=...,
                      function_name=..., code_snippet=...),
    sarif_rule_id=f"sec-af/{finding.finding_type.value}/{finding.cwe_id.lower()}",
    sarif_security_severity=0.0)

Two things distinguish it from agents/prove.Fallback, which serves the same purpose elsewhere:

  • the rationale is a FIXED string, not "Verification incomplete: <reason>";
  • the sarif_rule_id is built from the lowercased CWE ID ("sec-af/sast/cwe-89"), where prove's is built from the CWE NAME with spaces turned into hyphens ("sec-af/sast/sql-injection").

Python parity: nothing in the Python tree calls this function — every demotion path goes through agents/prove.fallback — so it is ported for completeness. `related_locations`, `compliance` and `reproduction_steps` keep pydantic's empty-list defaults; `drop_reason` stays None, which is what makes a finding built here invisible to the orchestrator's drop_reason sweep.

Types

type AttributeError 

type AttributeError struct{ Attr string }

AttributeError reproduces the CPython AttributeError a phase gets when it reaches for an attribute `_PhaseHarnessProxy` does not define. The message is byte-identical to CPython's, which matters: agents/prove's demotion classifier (`demoteOnError`) inspects `str(exc)` for "unverified"/"verdict"/ "validationerror", and this text matches none of them — so a verdict call through the proxy demotes with drop_reason "verifier_error", exactly as it does in Python.

func (*AttributeError) Error 

func (e *AttributeError) Error() string

type AuditOrchestrator 

type AuditOrchestrator struct {
	// App is the SDK agent. Python stores it as `cast("Any", app)` purely to
	// silence the type checker; the capability set it actually uses is
	// harness + ai + note (never call — the orchestrator's own pipeline is
	// in-process).
	App appx.App
	// Input is the audit request.
	Input schemas.AuditInput
	// StartedAt is `time.monotonic()` at construction. Go's time.Now is a wall
	// clock, but every read is a DIFFERENCE (`time.monotonic() - self.started_at`)
	// and Go's time.Time carries a monotonic reading that Sub prefers, so the
	// elapsed values match Python's semantics.
	StartedAt time.Time
	// RepoPath is `Path(os.getenv("SEC_AF_REPO_PATH", os.getcwd())).resolve()`.
	// app.py replaces it right after construction — see SetRepoPath.
	RepoPath string
	// CheckpointDir is `self.repo_path / ".sec-af"`.
	CheckpointDir string
	// IsPRMode is `input.is_pr`.
	IsPRMode bool
	// DiffAnalysis is populated only in PR mode with a base commit; nil is
	// Python's None.
	DiffAnalysis *diffanalysis.DiffAnalysis
	// Config is `AuditConfig.from_input(input, str(repo_path))`.
	Config config.AuditConfig
	// BudgetConfig is `self.config.budget`.
	BudgetConfig config.BudgetConfig
	// MaxCostUSD is `input.max_cost_usd` (None => nil => no cost budget).
	MaxCostUSD *float64
	// MaxDurationSeconds is `input.max_duration_seconds`.
	MaxDurationSeconds *int
	// FindingsNotVerified is the count GenerateOutput reports in metadata.
	//
	// Unlike the two guarded counters it stays an EXPORTED field, because
	// every writer runs on the sequential audit path and never inside the
	// phase fan-out: run.go's two phases, and app.py's own
	// `orchestrator.findings_not_verified = ...` (node/audit.go).
	FindingsNotVerified int
	// ProveDropSummary is the `{"demoted_total", "by_reason", "findings"}`
	// dict, rebuilt at the start of each prove pass and copied verbatim into
	// the result metadata.
	//
	// Typed `any`, not `map[string]any`, because on the app.py path it is
	// assigned straight from a `.call` payload with
	// `prove_dict.get("drop_summary", {...})` — a `.get`, so the default fires
	// ONLY when the key is absent. A key that is present with an odd value (a
	// JSON null, a string, a list) is threaded through to
	// `metadata["prove_drop_summary"]` verbatim in Python, and a
	// `map[string]any` field cannot hold it. See node/audit.go.
	ProveDropSummary any
	// AIGate is `AIGateWrapper(app=self.app)` — used by
	// AssessReachabilityParallel and by GenerateOutput's compliance fallback.
	AIGate *gates.AIGate
	// contains filtered or unexported fields
}

AuditOrchestrator ports the class of the same name (orchestrator.py:53).

Every field Python assigns in `__init__` is here, exported where a caller (app.py, and therefore internal/secaf/audit) reads or writes it: 

self.app, self.input, self.started_at, self.repo_path, self.checkpoint_dir,
self.is_pr_mode, self.diff_analysis, self.config, self.budget_config,
self.max_cost_usd, self.max_duration_seconds, self.total_cost_usd,
self.cost_breakdown, self.agent_invocations, self.budget_exhausted,
self.findings_not_verified, self.prove_drop_summary, self.ai_gate

func New 

func New(app appx.App, input schemas.AuditInput) (*AuditOrchestrator, error)

New ports `AuditOrchestrator.__init__(app, input)`. 

self.started_at = time.monotonic()
self.repo_path = Path(os.getenv("SEC_AF_REPO_PATH", os.getcwd())).resolve()
self.checkpoint_dir = self.repo_path / ".sec-af"
self.is_pr_mode = input.is_pr
if self.is_pr_mode and input.base_commit_sha:
    self.diff_analysis = analyze_diff(str(self.repo_path), input.base_commit_sha, input.commit_sha or "HEAD")
self.config = AuditConfig.from_input(self.input, str(self.repo_path))
...
self.ai_gate = AIGateWrapper(app=self.app)

Python parity:

  • the diff analysis runs against the CONSTRUCTOR's repo path, before app.py substitutes the resolved (possibly cloned) one. Reproduced: callers that want the diff against the real checkout must construct after resolving, which app.py does not do.
  • `input.base_commit_sha` and `input.commit_sha` are tested with PYTHON TRUTHINESS, so an empty string behaves like None — no diff analysis, and the head defaults to "HEAD".
  • AIGateWrapper's config falls back to `AIIntegrationConfig.from_env()`; here it is config.DefaultAIConfig(), which reads nothing and cannot fail.

Signature parity note: Python's `__init__` is synchronous and carries no context, so New takes none. analyze_diff shells out to git, and diffanalysis.AnalyzeDiff gives each invocation its own 30s timeout, so context.Background() is the honest translation. NewWithContext is for a caller that already holds a cancellable context.

func NewAt 

func NewAt(ctx context.Context, app appx.App, input schemas.AuditInput, repoPath string) (*AuditOrchestrator, error)

NewAt is NewWithContext for a repository the caller has already resolved to an absolute directory, which is how codeaf's audit entry (internal/secaf/audit) constructs it.

It exists because the Python order — construct against SEC_AF_REPO_PATH or the process working directory, then overwrite repo_path — runs the PR-mode diff analysis and records AuditConfig.repo_path against whatever directory codeaf happened to start in. Handing the repository in up front makes all three (repo_path, checkpoint_dir, config.repo_path) and the diff analysis agree on the repository being audited.

func NewWithContext 

func NewWithContext(ctx context.Context, app appx.App, input schemas.AuditInput) (*AuditOrchestrator, error)

NewWithContext is New with the context the PR-mode git commands run under.

func (*AuditOrchestrator) AgentInvocations 

func (o *AuditOrchestrator) AgentInvocations() int

AgentInvocations reports `self.agent_invocations` — the number of harness invocations made through the phase proxy.

func (*AuditOrchestrator) AssessReachabilityParallel 

func (o *AuditOrchestrator) AssessReachabilityParallel(ctx context.Context, verified []schemas.VerifiedFinding)

AssessReachabilityParallel ports `_assess_reachability_parallel(verified)` (orchestrator.py:568): 

needs_assessment = [f for f in verified if not any(tag in reachability_tags for tag in f.tags)]
if not needs_assessment: return
semaphore = asyncio.Semaphore(min(5, len(needs_assessment)))
async def _assess_one(finding):
    async with semaphore:
        try:
            gate_result = await self.ai_gate.assess_reachability(summary)
            finding.tags.append(gate_result.reachability)
        except Exception:
            finding.tags.append("requires_auth")   # safe default
await asyncio.gather(*[_assess_one(f) for f in needs_assessment])

Python parity:

  • the semaphore is `min(5, len(needs_assessment))`, NOT max(1, ...): the early return above guarantees the count is at least 1, so the limit is between 1 and 5.
  • EVERY failure — a gate error, a malformed response, anything — appends the literal "requires_auth". That is the documented safe default: an unassessable finding is treated as authenticated-only rather than internet-facing, which lowers its reachability multiplier in scoring.
  • the gate's `reachability` string is appended VERBATIM, with no validation. A gate that answers "maybe" puts "maybe" in the tags, and scoring.reachabilityMultiplier then falls through to its default.
  • findings are MUTATED IN PLACE. The Go port takes the slice and indexes into it, so the caller sees the tags without a return value — the same aliasing Python has.

Concurrency: each goroutine appends to a DIFFERENT element's Tags slice, so no lock is needed around the append itself; the WaitGroup provides the happens-before edge the caller needs to read them.

func (*AuditOrchestrator) BudgetExhausted 

func (o *AuditOrchestrator) BudgetExhausted() bool

BudgetExhausted reports `self.budget_exhausted`, which latches true the first time a budget check trips.

func (*AuditOrchestrator) BuildProgress 

func (o *AuditOrchestrator) BuildProgress(phase string, agentsTotal, agentsCompleted, findingsSoFar int) schemas.AuditProgress

BuildProgress is EmitProgress without the note — the AuditProgress value it serializes. Split out so a test can assert the arithmetic without scripting a fake, and so the note message can be golden-compared.

func (*AuditOrchestrator) CheckpointPath 

func (o *AuditOrchestrator) CheckpointPath(phase string) string

CheckpointPath ports `_checkpoint_path(phase)`: 

return self.checkpoint_dir / f"checkpoint-{phase}.json"

func (*AuditOrchestrator) CostBreakdown 

func (o *AuditOrchestrator) CostBreakdown() map[string]float64

CostBreakdown returns a COPY of the per-phase cost map (`self.cost_breakdown`), so a caller cannot corrupt the running totals.

func (*AuditOrchestrator) DefaultStrategies 

func (o *AuditOrchestrator) DefaultStrategies(recon schemas.ReconResult) []schemas.HuntStrategy

DefaultStrategies ports `AuditOrchestrator._default_strategies(recon)` (orchestrator.py:507).

This is the ORCHESTRATOR's variant and it is NOT the same list as reasoners/phases.py `_default_strategies(recon, depth)`: 

base:             injection, dos, ssrf, auth, data_exposure, config_secrets
+ 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
+ python_specific      when depth is thorough and "python" in the lowered languages
+ javascript_specific  when depth is thorough and any language is javascript/typescript

Differences from the phases copy, both deliberate: XSS is NEVER added here (phases adds it at standard/thorough), and the two language-specific strategies exist only here.

Python parity:

  • the depth comes from `self._depth_profile()`, i.e. the orchestrator's own input, not a parameter;
  • `"python" in {lang.lower() for lang in recon.languages}` is an EXACT match on the lowered name — "python3" does not qualify;
  • the javascript test uses `any(lang.lower() in {"javascript", "typescript"})`, so either language alone adds JAVASCRIPT_SPECIFIC, and both together add it once;
  • the trailing pass keeps FIRST-seen order and drops repeats. No branch can add a duplicate today, but it is reproduced.

func (*AuditOrchestrator) EmitProgress 

func (o *AuditOrchestrator) EmitProgress(ctx context.Context, phase string, agentsTotal, agentsCompleted, findingsSoFar int)

EmitProgress ports `_emit_progress` (orchestrator.py:636): 

elapsed = time.monotonic() - self.started_at
safe_total = max(1, agents_total)
phase_progress = min(1.0, agents_completed / safe_total)
estimated_total = elapsed / phase_progress if phase_progress > 0 else elapsed
progress = AuditProgress(
    phase=phase, phase_progress=phase_progress,
    agents_total=agents_total, agents_completed=agents_completed,
    agents_running=max(0, agents_total - agents_completed),
    findings_so_far=findings_so_far, elapsed_seconds=elapsed,
    estimated_remaining_seconds=max(0.0, estimated_total - elapsed),
    cost_so_far_usd=round(self.total_cost_usd, 4))
self.app.note(progress.model_dump_json(), tags=["audit", "progress", phase])

Python parity:

  • `agents_completed / safe_total` is TRUE division, so 1/2 is 0.5, not 0;
  • `safe_total = max(1, agents_total)` guards a zero total, and the outer min clamps a completed count that overshoots;
  • `estimated_total` divides by phase_progress, so a phase with zero progress reports an estimated remaining of 0.0 rather than infinity;
  • `round(self.total_cost_usd, 4)` is Python's BANKER'S rounding (pyfmt.Round), not Go's strconv default.

The note MESSAGE is `progress.model_dump_json()` — pydantic's serializer, not json.dumps: no space after `:` or `,`, and a float field always carries a decimal point (`"phase_progress":1.0`). pyfmt.DumpsModelJSON is that spelling; a Go `json.Marshal` would emit `1` and break byte parity with the Python node's notes.

func (*AuditOrchestrator) GenerateOutput 

func (o *AuditOrchestrator) GenerateOutput(
	ctx context.Context,
	recon schemas.ReconResult,
	hunt schemas.HuntResult,
	verified []schemas.VerifiedFinding,
) (schemas.SecurityAuditResult, error)

GenerateOutput ports `_generate_output(recon, hunt, verified)` (orchestrator.py:365) — the tail every entry point shares. app.py calls it directly after its four `.call`s; run() and run_from_checkpoint() call it last.

Steps, in Python's order:

  1. THRESHOLD FILTER. `severity_order.get(self.input.severity_threshold.lower(), 0)`; when that is > 0, drop every finding scoring below it. A threshold of "info" or an unknown string scores 0 and filters NOTHING.
  2. PER-FINDING ENRICHMENT, in place: the CWE severity floor, the exploitability score, sarif_security_severity mirrored from it, and the compliance mappings (static table, with the AI gate as fallback).
  3. COUNTS. verdict counts over the four Verdict members; severity counts seeded with every Severity member so a severity with no findings still reports 0.
  4. NOISE REDUCTION. `not_exploitable / hunt.total_raw * 100`, rounded to 2 with Python's banker's rounding. Guarded on total_raw > 0.
  5. ATTACK CHAINS. PotentialChain -> AttackChain, with `description` taking the chain's combined_impact (so description and combined_impact are the same string) and mitre_attack_mapping left at None.
  6. COMPLIANCE GAPS, the SecurityAuditResult, then SARIF, then the JSON and Markdown reports whose return values are DISCARDED (`_ = generate_json(...)`) — they are produced for their side-effect-free validation only.
  7. The compliance-report loop is NOT carried: see the end of the function.

Python parity details worth stating:

  • `verified` is REBOUND by the filter, so the caller's slice is not truncated — but the surviving findings are the SAME objects and step 2 mutates them. Go reproduces both: a new slice header, sharing elements.
  • `self.input.compliance_frameworks or None` passes None for an EMPTY list, which makes get_compliance_mappings_hybrid use its default framework set.
  • `commit_sha=self.input.commit_sha or "HEAD"` is Python truthiness: an empty string also becomes "HEAD".
  • `timestamp=datetime.now(UTC)` is read here, not at construction.
  • `duration_seconds` is `time.monotonic() - self.started_at`, i.e. the whole orchestrator lifetime, not just this function.
  • the budget note fires only when the flag is latched, and reports findings_not_verified — which app.py copies from prove_phase's payload.

func (*AuditOrchestrator) MergeRecon 

func (o *AuditOrchestrator) MergeRecon(
	fast schemas.ReconResult,
	dataFlows schemas.DataFlowMap,
	securityContext schemas.SecurityContext,
) schemas.ReconResult

MergeRecon ports `_merge_recon(fast, deep_result)` (orchestrator.py:196): 

data_flows, security_context = deep_result
frameworks = sorted({item for item in security_context.framework_security if item})
return ReconResult(architecture=fast.architecture, data_flows=data_flows,
                   dependencies=fast.dependencies, config=fast.config,
                   security_context=security_context, languages=fast.languages,
                   frameworks=frameworks, lines_of_code=fast.lines_of_code,
                   file_count=fast.file_count)

Python parity: `recon_duration_seconds` is NOT carried over from `fast` — the merged result keeps the pydantic default 0.0 — and the frameworks come from the DEEP security context, not the fast one.

func (*AuditOrchestrator) PhaseProxy 

func (o *AuditOrchestrator) PhaseProxy(phase string) appx.Harnesser

PhaseProxy exposes the budget-checking wrapper for one phase name ("recon", "hunt" or "prove"). It is exported so a caller outside this package can drive a phase with the same accounting the orchestrator applies.

The return type is appx.Harnesser, not appx.App: see the phaseProxy doc.

func (*AuditOrchestrator) PrioritizeFindings 

func (o *AuditOrchestrator) PrioritizeFindings(findings []schemas.RawFinding) []schemas.RawFinding

PrioritizeFindings ports `AuditOrchestrator._prioritize_findings` (orchestrator.py:539) — severity first, confidence second, both descending, unknown values scoring 0.

It is byte-identical to reasoners/phases.py's `_prioritize_findings` and to agents/prove's `_priority_sort`; SEC-AF carries three copies and each Go package ports its own.

Python parity: `sorted(..., reverse=True)` is STABLE (CPython reverses, sorts, reverses), so ties keep input order — sort.SliceStable with a strictly-greater comparison. The input slice is not mutated.

func (*AuditOrchestrator) ProvePhaseProxy 

func (o *AuditOrchestrator) ProvePhaseProxy() proveagent.HarnessAIer

ProvePhaseProxy is PhaseProxy(PhaseProve) with the AI seam described above.

func (*AuditOrchestrator) ProverCap 

func (o *AuditOrchestrator) ProverCap() int

ProverCap ports `AuditOrchestrator._prover_cap()` (orchestrator.py:557): 

defaults = {QUICK: 10, STANDARD: 30, THOROUGH: 10_000}
default_cap = defaults[self._depth_profile()]
if self.input.max_provers is None: return default_cap
return max(0, min(self.input.max_provers, default_cap))

Same arithmetic as phases' `_prover_cap`, but the depth and max_provers come from the orchestrator's own input rather than from arguments.

func (*AuditOrchestrator) Run 

func (o *AuditOrchestrator) Run(ctx context.Context) (schemas.SecurityAuditResult, error)

Run ports `AuditOrchestrator.run()` (orchestrator.py:82) — the STREAMING, fully in-process pipeline. 

note("Starting SEC-AF streaming orchestrator", tags=["audit","start","streaming"])
fast_recon = await self._run_fast_recon();  self._write_checkpoint("recon_fast", fast_recon)
findings_queue = asyncio.Queue()
deep_result, hunt, verified = await asyncio.gather(
    self._run_deep_recon_async(fast_recon),
    self._run_hunt_streaming(fast_recon, findings_queue),
    self._run_prove_streaming(findings_queue))
self.findings_not_verified = max(0, len(hunt.findings) - len(verified))
recon = self._merge_recon(fast_recon, deep_result)
write "recon", "hunt", "prove"
await self._assess_reachability_parallel(verified)
rebuild prove_drop_summary from each finding's drop_reason
(DAST branch — dead, see enableDast)
result = await self._generate_output(...)
note("SEC-AF audit complete", tags=["audit","complete"])

app.py does NOT use this path: it issues four `.call`s into internal/phases and then calls GenerateOutput directly. Run is ported because RunFromCheckpoint shares its `_run_hunt`/`_run_prove` halves and because DESIGN.md §3 requires the whole class.

Concurrency parity: the three tasks run together, the hunt publishing batches to the prove consumer through a queue. Go uses one errgroup (no WithContext — `asyncio.gather` does not cancel siblings on failure either), so the FIRST error is returned once all three have finished. The Go channel is buffered to the strategy count and CLOSED by RunHuntStreaming in place of Python's None sentinel.

func (*AuditOrchestrator) RunDASTVerification 

func (o *AuditOrchestrator) RunDASTVerification(ctx context.Context, verified []schemas.VerifiedFinding)

RunDASTVerification ports `_run_dast_verification(verified)` (orchestrator.py:332). 

confirmed = [f for f in verified if f.verdict == Verdict.CONFIRMED]
if not confirmed:
    note("No confirmed findings available for DAST step", tags=["audit","prove","dast"]); return
for finding in confirmed:
    try: dast_result = await run_dast_verifier(proxy, str(self.repo_path), finding)
    except Exception as exc:
        finding.tags.append("dast_error")
        note(f"DAST verifier failed for '{finding.title}': {exc}", tags=["audit","prove","dast","error"])
        continue
    finding.tags.append("dast_attempted" if dast_result.exploit_attempted else "dast_skipped")
    finding.tags.append("dast_confirmed" if dast_result.exploit_succeeded else "dast_not_confirmed")
    finding.rationale = f"{finding.rationale}\nDAST: {dast_result.response_analysis}"
    if finding.proof is not None:
        finding.proof.poc_execution_output = json.dumps({...}, indent=2)

UNREACHABLE IN PRODUCTION on two independent counts — the enableDast guard and ErrDastVerifierArity — both documented above. With the production seam every confirmed finding takes the except branch: tagged "dast_error", noted, and otherwise untouched.

Findings are mutated IN PLACE through the shared backing array, matching Python's aliasing.

func (*AuditOrchestrator) RunDeepReconAsync 

func (o *AuditOrchestrator) RunDeepReconAsync(ctx context.Context, fastRecon schemas.ReconResult) (
	schemas.DataFlowMap, schemas.SecurityContext, error,
)

RunDeepReconAsync ports `_run_deep_recon_async(fast_recon)` (orchestrator.py:182): 

if self.config.depth == DepthProfile.QUICK:
    return fast_recon.data_flows, fast_recon.security_context
result = await run_deep_recon(app=proxy, repo_path=..., architecture=fast_recon.architecture)
self._emit_progress(phase="recon", agents_total=2, agents_completed=2, findings_so_far=0)
return result

Python parity: the QUICK short circuit reads `self.config.depth`, which AuditConfig.from_input already normalized — NOT `self._depth_profile()`. The two agree, but the config is the one the code consults. It also emits NO progress event, so a quick audit reports recon 1/2 and never 2/2.

func (*AuditOrchestrator) RunFastRecon 

func (o *AuditOrchestrator) RunFastRecon(ctx context.Context) (schemas.ReconResult, error)

RunFastRecon ports `_run_fast_recon()` (orchestrator.py:165) — the three cheap mappers, plus the PR-mode cache short circuit.

Python parity: the progress event reports agents_total=2 / agents_completed=1 on BOTH paths (cached and fresh), because the deep half is still to come.

func (*AuditOrchestrator) RunFromCheckpoint 

func (o *AuditOrchestrator) RunFromCheckpoint(ctx context.Context, phase string) (schemas.SecurityAuditResult, error)

RunFromCheckpoint ports `run_from_checkpoint(phase)` (orchestrator.py:121): 

normalized_phase = phase.lower().strip()
if normalized_phase not in {"recon", "hunt", "prove"}:
    raise ValueError(f"Unknown checkpoint phase: {phase}")
recon:  always read from checkpoint-recon.json
"recon": run hunt, write it, run prove, write it
"hunt":  read hunt, run prove, write it
"prove": read hunt AND the prove list — nothing is re-run
return await self._generate_output(recon=recon, hunt=hunt, verified=verified)

Python parity:

  • the error message carries the ORIGINAL phase string, not the normalized one, so `run_from_checkpoint(" Recon ")` succeeds while `run_from_checkpoint("Recon!")` reports "Unknown checkpoint phase: Recon!".
  • normalization is `.lower()` THEN `.strip()`; both orders agree for whitespace, and Go's TrimSpace(ToLower(s)) matches for every input that is not exotic Unicode whitespace.
  • the "prove" branch reads a LIST of VerifiedFinding, the other two read single models — hence the two read helpers.

func (*AuditOrchestrator) RunHunt 

func (o *AuditOrchestrator) RunHunt(ctx context.Context, recon schemas.ReconResult) (schemas.HuntResult, error)

RunHunt ports `_run_hunt(recon)` (orchestrator.py:263) — the non-streaming hunt used by run_from_checkpoint("recon").

func (*AuditOrchestrator) RunHuntStreaming 

func (o *AuditOrchestrator) RunHuntStreaming(
	ctx context.Context,
	recon schemas.ReconResult,
	findingsQueue chan<- []schemas.RawFinding,
) (schemas.HuntResult, error)

RunHuntStreaming ports `_run_hunt_streaming(recon, findings_queue)` (orchestrator.py:211) — run_hunt_streaming plus the recon-finding merge and the progress event.

The channel is CLOSED by agents/hunt.RunHuntStreaming in place of Python's None sentinel, including when a hunter fails, so the prove consumer always terminates.

func (*AuditOrchestrator) RunProve 

func (o *AuditOrchestrator) RunProve(
	ctx context.Context,
	recon schemas.ReconResult,
	hunt schemas.HuntResult,
) ([]schemas.VerifiedFinding, error)

RunProve ports `_run_prove(recon, hunt)` (orchestrator.py:290) — the non-streaming prove used by run_from_checkpoint. 

prioritized = self._prioritize_findings(hunt.findings)
limited_hunt = HuntResult(findings=prioritized[:self._prover_cap()], <everything else copied>)
self.findings_not_verified = max(0, len(hunt.findings) - len(limited_hunt.findings))
verified = await run_prove(app=proxy, ...)
await self._assess_reachability_parallel(verified)
rebuild prove_drop_summary
(DAST branch — dead)
self._emit_progress(phase="prove", agents_total=1, agents_completed=1, findings_so_far=len(verified))

Python parity: `recon` is accepted and immediately discarded (`_ = recon`), and the limited hunt copies chains/total_raw/deduplicated_count/chain_count/ strategies_run/hunt_duration_seconds VERBATIM — only the findings list is truncated, so total_raw keeps counting the findings that were dropped.

func (*AuditOrchestrator) RunProveStreaming 

func (o *AuditOrchestrator) RunProveStreaming(
	ctx context.Context,
	findingsQueue <-chan []schemas.RawFinding,
) []schemas.VerifiedFinding

RunProveStreaming ports `_run_prove_streaming(findings_queue)` (orchestrator.py:244).

Python parity: it RESETS findings_not_verified to 0 before consuming, so the value run() computes afterwards (`len(hunt.findings) - len(verified)`) is the one that survives. run_prove_streaming itself cannot fail, so neither can this.

func (*AuditOrchestrator) RunRecon 

func (o *AuditOrchestrator) RunRecon(ctx context.Context) (schemas.ReconResult, error)

RunRecon ports `_run_recon()` (orchestrator.py:149) — the non-streaming RECON phase.

Python parity: NOTHING calls it. run() uses _run_fast_recon plus _run_deep_recon_async, and run_from_checkpoint always reads recon from disk. It is ported for completeness, and it is the only place the PR-mode recon cache is consulted with the "Using cached recon for PR-mode scan" wording (note the trailing " scan", which _run_fast_recon's otherwise-identical note lacks).

func (*AuditOrchestrator) SetAgentInvocations 

func (o *AuditOrchestrator) SetAgentInvocations(n int)

SetAgentInvocations OVERWRITES the counter, which is what app.py:220 does (`orchestrator.agent_invocations = total_selected + len(...) + 3`) before calling _generate_output.

func (*AuditOrchestrator) SetRepoPath 

func (o *AuditOrchestrator) SetRepoPath(repoPath string)

SetRepoPath performs the two assignments app.py makes right after construction: 

orchestrator.repo_path = Path(repo_path)
orchestrator.checkpoint_dir = orchestrator.repo_path / ".sec-af"

Python parity: app.py does NOT re-resolve the path here (it passes the already-absolute result of _resolve_repo) and does NOT recompute self.config, so AuditConfig.repo_path keeps the constructor's value. Both are reproduced — this setter touches exactly the two fields Python touches.

func (*AuditOrchestrator) TotalCostUSD 

func (o *AuditOrchestrator) TotalCostUSD() float64

TotalCostUSD reports the accumulated harness cost (`self.total_cost_usd`).

func (*AuditOrchestrator) TrackDrop 

func (o *AuditOrchestrator) TrackDrop(ctx context.Context, findingTitle string, originalVerdict *string, reason string)

TrackDrop ports `AuditOrchestrator._track_drop` (orchestrator.py:654) — the demotion bookkeeping written into prove_drop_summary, plus its note.

It is the same bookkeeping as reasoners/phases.py's module-level `_track_drop` with ONE difference that is easy to miss: the note tags here are ["audit", "prove", "drop"], where phases uses ["prove", "drop", "demotion"].

Python parity:

  • `setdefault` recreates a missing "by_reason"/"findings" entry, which the Go type assertions reproduce;
  • the recorded `original_verdict` is None (JSON null) when absent;
  • the note's `original_verdict or 'unknown'` is PYTHON TRUTHINESS, so an empty string also prints "unknown".

func (*AuditOrchestrator) TryLoadCachedRecon 

func (o *AuditOrchestrator) TryLoadCachedRecon() *schemas.ReconResult

TryLoadCachedRecon ports `_try_load_cached_recon()`: 

try:
    return self._read_checkpoint("recon", ReconResult)
except (FileNotFoundError, Exception):
    return None

Python parity: the except tuple is redundant — `Exception` already covers FileNotFoundError — so EVERY failure (missing file, malformed JSON, failed validation) yields None. A nil pointer is that None.

func (*AuditOrchestrator) WriteCheckpoint 

func (o *AuditOrchestrator) WriteCheckpoint(phase string, payload any) error

WriteCheckpoint ports `_write_checkpoint(phase, payload)`: 

self.checkpoint_dir.mkdir(parents=True, exist_ok=True)
path = self._checkpoint_path(phase)
data = [item.model_dump() for item in payload] if isinstance(payload, list) else payload.model_dump()
body = {"phase": phase, "created_at": datetime.now(UTC).isoformat(), "data": data}
path.write_text(json.dumps(body, indent=2), encoding="utf-8")

The FILE BYTES are the contract — a Go node must be able to resume from a checkpoint a Python node wrote and vice versa — so the body is rendered with pyfmt.Dumps (json.dumps parity: `", "`/`": "` separators become `,`/`": "` under indent, floats keep Python's repr, non-ASCII is \uXXXX-escaped and `<>&` are NOT escaped) over an INSERTION-ORDERED object, because a Go map would sort "created_at" before "data" and "phase" while Python emits them in the literal's order.

Python parity:

  • `created_at` is a plain STRING produced by `datetime.now(UTC).isoformat()` — `2026-01-02T03:04:05.123456+00:00`, with the fraction omitted when the microseconds are zero. schemas.Timestamp.String() is that exact spelling.
  • the isinstance(list) branch is only about calling model_dump per element; pyfmt.Dumps renders a slice of structs as an array of objects with the same field order, so one code path covers both.
  • Python raises on an I/O failure; Go returns the error, and every caller propagates it.
  • no trailing newline is written (`json.dumps` produces none, and `write_text` adds none).

type BudgetExhaustedError 

type BudgetExhaustedError struct{ Message string }

BudgetExhaustedError ports `class BudgetExhausted(RuntimeError)` (orchestrator.py:35) — the sentinel the phase proxy raises when a cost or duration budget is used up.

func (*BudgetExhaustedError) Error 

func (e *BudgetExhaustedError) Error() string

type UnknownCheckpointPhaseError 

type UnknownCheckpointPhaseError struct{ Phase string }

UnknownCheckpointPhaseError is `ValueError(f"Unknown checkpoint phase: {phase}")`.

app.py maps a ValueError to HTTP 400, so the type matters as much as the text: `except ValueError as exc: raise HTTPException(400, {"error": str(exc)})`.

func (*UnknownCheckpointPhaseError) Error 

func (e *UnknownCheckpointPhaseError) Error() string

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.