gates

func BuildAIIntegration ¶

func BuildAIIntegration(app appx.App, cfg *config.AIIntegrationConfig) (*HarnessWrapper, *AIGate, error)

BuildAIIntegration ports harness.py build_ai_integration:

resolved = config or AIIntegrationConfig.from_env()
return HarnessWrapper(app=app, config=resolved), AIGateWrapper(app=app, config=resolved)

Both wrappers share ONE resolved config instance in Python; they share one value here, with config.DefaultAIConfig() standing in for from_env. Nothing in the Python tree calls this function — orchestrator.py constructs AIGateWrapper directly — so it exists for completeness.

func BuildSchemaRetryPrompt ¶

func BuildSchemaRetryPrompt[T any](errorDetail, cwd string) string

BuildSchemaRetryPrompt ports harness.py _build_schema_retry_prompt:

output_path = Path(cwd) / ".agentfield_output.json"
schema_json_str = json.dumps(schema.model_json_schema(), indent=2)
return (f"The JSON output at {output_path} failed validation.\n"
        f"Error: {error_detail}\n\n"
        f"Your response must conform to this JSON schema:\n"
        f"```json\n{schema_json_str}\n```\n\n"
        f"Rewrite the COMPLETE, corrected JSON to: {output_path}\n"
        f"The file must contain ONLY valid JSON matching the schema above. "
        f"No markdown fences, no extra text, no comments.")

The embedded schema JSON is byte-reproducible from the committed fixture. The fixture is written with sort_keys=True, and pydantic's own model_json_schema() dict happens to be in sorted key order at EVERY level except one: the `properties` object of the root model and of each `$defs` entry, which are in field DECLARATION order. declarationOrdered restores exactly those, using the Go struct field order reachable from T — verified against all 23 committed fixtures.

func HarnessInvoke ¶

func HarnessInvoke[T any](ctx context.Context, w *HarnessWrapper, req InvokeRequest) (T, error)

HarnessInvoke ports HarnessWrapper.invoke.

self._cost_tracker.register_invocation()
enhanced_prompt = f"{_with_phase_guidance(prompt, phase, cwd)}\n\n{_schema_guidance(schema)}"
result = await self._run_with_retry(_operation, self.config)
self._cost_tracker.register_cost(result.cost_usd)
if result.is_error: raise AIIntegrationError(f"Harness failure{f' ({phase})' if phase else ''}: {message}")
if isinstance(parsed, schema): return parsed
...
if not parsed and not result.is_error: return await self._invoke_with_schema_retry(...)
raise AIIntegrationError(f"Harness returned invalid payload for schema {schema.__name__}")

Python parity notes:

• The retry wrapper covers the SDK CALL only. A harness that RAN and failed comes back with is_error set and a nil Go error, so it never reaches _run_with_retry's except clause — it goes straight to the AIIntegrationError raise, transient message or not. That is Python's behavior too.
• The phase suffix on the failure message is the empty string when phase is falsy, so an empty phase contributes nothing to the text, not " ()".
• The `isinstance(result, schema)` third branch is Python duck-typing for an SDK that returns the model itself; the Go SDK always returns a *appx.HarnessResult, so it is unreachable and not ported.

func Invoke ¶

func Invoke[T any](ctx context.Context, g *AIGate, user, system string) (T, error)

Invoke ports AIGateWrapper.invoke:

self._cost_tracker.register_invocation()
async def _operation():
    return await self.app.ai(system=system, user=user, schema=schema, model=self.config.ai_model)
result = await self._run_with_retry(_operation, self.config)
self._cost_tracker.register_cost(getattr(result, "cost_usd", None))
if isinstance(result, schema): return result
if isinstance(result, dict):   return schema(**result)
raise AIIntegrationError(f"AI gate returned invalid payload for schema {schema.__name__}")

It is a package-level function rather than a method because Go methods cannot carry their own type parameter, and the schema is exactly that.

Python parity notes:

• register_invocation fires ONCE per Invoke, before the first attempt — a retried call still counts as one invocation.
• `system=None` is spelled as the empty string here; aix.Structured omits the system message for it, matching the Python SDK, which drops a None system.
• The two isinstance branches collapse into aix.Structured's typed return: the Go SDK hands back raw text, aix parses it into T, and a parse failure is an error — the same outcome as the trailing `raise`. The dict branch exists in Python only because the SDK may hand back an unparsed dict.
• COST: this is a DELIBERATE, design-doc-mandated divergence. Python reads `getattr(result, "cost_usd", None)` off the PARSED PYDANTIC MODEL the SDK returns for a `schema=` call (agent_ai.py:898 `return schema(**json_data)`), and no gate schema has a cost_usd field — so AIGateWrapper.total_cost_usd is permanently 0.0 in the Python node. DESIGN.md §2/§3 specifies the Go port take the cost from the SDK response's Usage.Cost instead, which is where the real number lives. Everything else about the tracker (invocation counting, the negative/None guard, the += accumulation) is byte-for-byte Python.

func IsTransientError ¶

func IsTransientError(errText string) bool

IsTransientError ports harness.py _is_transient_error:

lowered = error.lower()
return any(pattern in lowered for pattern in _TRANSIENT_PATTERNS)

func RunHuntAnalysis ¶

func RunHuntAnalysis[T any](ctx context.Context, w *HarnessWrapper, prompt, cwd, projectDir string) (T, error)

RunHuntAnalysis ports HarnessWrapper.run_hunt_analysis — invoke with phase="hunt".

func RunProveAnalysis ¶

func RunProveAnalysis[T any](ctx context.Context, w *HarnessWrapper, prompt, cwd, projectDir string) (T, error)

RunProveAnalysis ports HarnessWrapper.run_prove_analysis — invoke with phase="prove".

func RunReconAnalysis ¶

func RunReconAnalysis[T any](ctx context.Context, w *HarnessWrapper, prompt, cwd, projectDir string) (T, error)

RunReconAnalysis ports HarnessWrapper.run_recon_analysis — invoke with phase="recon".

func SchemaGuidance ¶

func SchemaGuidance[T any]() string

SchemaGuidance ports harness.py _schema_guidance:

field_lines = []
for field_name, field in schema.model_fields.items():
    description = (field.description or "").strip()
    if description:
        field_lines.append(f"- `{field_name}`: {description}")
if not field_lines:
    return ("Output format:\n"
            "- Return valid JSON only (no markdown fences, no extra text).\n"
            "- Follow the provided Pydantic schema exactly.")
return ("Output format:\n"
        "- Return valid JSON only (no markdown fences, no extra text).\n"
        "- Follow the provided Pydantic schema exactly.\n"
        "- Field guidance from schema descriptions:\n"
        f"{chr(10).join(field_lines)}")

The two Python inputs — the field ORDER and each field's DESCRIPTION — come from different places in Go:

• ORDER is `schema.model_fields.items()`, pydantic declaration order. The port's cross-package contract is that a Go struct's fields are declared in the same order as its pydantic counterpart's, so reflect.Type field order IS model_fields order. (The committed JSON-Schema fixture cannot supply it: go/scripts/gen_schemas.py writes with sort_keys=True, so its `properties` object is alphabetical.)
• DESCRIPTION is `FieldInfo.description`, which pydantic copies verbatim into `properties[<name>].description` of model_json_schema() — including for Optional and nested-model fields, where the description sits beside the anyOf/$ref rather than inside it. harnessx.SchemaFor[T] serves that document.

A field with no description contributes no line, exactly as in Python; a model with no described fields at all takes the short fallback.

func WithFileWriteHint ¶

func WithFileWriteHint(prompt, cwd string) string

WithFileWriteHint ports harness.py _with_file_write_hint:

output_path = Path(cwd) / ".agentfield_output.json"
return (f"{prompt.rstrip()}\n"
        f"- If output is large or complex, use the file-write pattern and ensure final JSON is written to {output_path}.")

Python parity notes:

• `str.rstrip()` with no argument strips ALL trailing whitespace, not just newlines; strings.TrimRight with unicode.IsSpace is the same set, and strings.TrimRightFunc(s, unicode.IsSpace) is what TrimSpace uses. Here strings.TrimRight over " \t\n\r\v\f" plus TrimRightFunc would be equivalent for every real prompt, so TrimRightFunc is used for exactness.
• `Path(cwd) / name` is NOT filepath.Join. pathlib collapses redundant separators and lone "." components but deliberately keeps "..", because resolving it without touching the filesystem would be wrong in the presence of symlinks; filepath.Join runs Clean, which eats "..". So Path("./work/../work")/"f" is "work/../work/f" where Join gives "work/f". pyPathJoin below reproduces pathlib. In the live node cwd is always an absolute tempfile.mkdtemp() path where the two agree, but the divergence is real and the golden test pins it.

func WithPhaseGuidance ¶

func WithPhaseGuidance(prompt, phase, cwd string) string

WithPhaseGuidance ports harness.py _with_phase_guidance:

normalized_phase = (phase or "").strip().lower()
phase_context = PHASE_GUIDANCE.get(normalized_phase, <default paragraph>)
constraints = _with_file_write_hint("Constraints:\n- ...\n- ...\n- ...", cwd)
return (f"Context:\n- {phase_context}\n\n"
        f"{constraints}\n\n"
        f"Task:\n{prompt.rstrip()}\n\n"
        "Output:\n"
        "- Return a single JSON object matching the requested schema.")

`phase` is `str | None` in Python; the Go signature takes a plain string because `(phase or "")` maps None and "" to the same normalized value.