feat(security): harness pattern extraction — prompt injection defense

CHANGELOG.md

@@ -15,6 +15,11 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- **Prompt-injection defense: `UntrustedData` wrapper.** All external content returned by `ReadFile` and `FetchURL` is now wrapped in `<untrusted_data id="NONCE">…</untrusted_data>` tags before being passed to the LLM, providing a clear boundary between trusted instructions and untrusted file/web content. The `UntrustedData` primitive escapes embedded closing tags to prevent breakout attacks.
+- **Agent boundary artifacts.** New `CodingArtifact` / `VerificationResult` and `VulnerabilityArtifact` / `AuditVerdict` frozen dataclasses in `pythinker_code.utils.artifacts` enforce a typed information barrier between coder and verifier subagents.
+- **Recon-first `planner` subagent.** A new read-only `planner` built-in agent type decomposes open-ended tasks into distinct parallel seed descriptions emitted as `<recon_seeds>` JSON, enabling structured fan-out before parallel workers start.
+- **Coder artifact contract.** The `coder` subagent now emits a `<coding_artifact>` JSON block at the end of every response, providing structured handoff data (`files_changed`, `test_command`, `expected_behavior`, optional `edge_cases_claimed`) that the `verifier` subagent can consume directly.
+
 ## 0.36.0 (2026-06-05)
 
 - **Alibaba DashScope multi-region fallback.** Logging in with a China-region key (`dashscope.aliyuncs.com`) against the default US Virginia endpoint now auto-detects the mismatch and reconfigures for the correct endpoint rather than failing with a misleading "API key is wrong" error.

docs/en/release-notes/changelog.md

@@ -17,6 +17,11 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- **Prompt-injection defense: `UntrustedData` wrapper.** All external content returned by `ReadFile` and `FetchURL` is now wrapped in `<untrusted_data id="NONCE">…</untrusted_data>` tags before being passed to the LLM, providing a clear boundary between trusted instructions and untrusted file/web content. The `UntrustedData` primitive escapes embedded closing tags to prevent breakout attacks.
+- **Agent boundary artifacts.** New `CodingArtifact` / `VerificationResult` and `VulnerabilityArtifact` / `AuditVerdict` frozen dataclasses in `pythinker_code.utils.artifacts` enforce a typed information barrier between coder and verifier subagents.
+- **Recon-first `planner` subagent.** A new read-only `planner` built-in agent type decomposes open-ended tasks into distinct parallel seed descriptions emitted as `<recon_seeds>` JSON, enabling structured fan-out before parallel workers start.
+- **Coder artifact contract.** The `coder` subagent now emits a `<coding_artifact>` JSON block at the end of every response, providing structured handoff data (`files_changed`, `test_command`, `expected_behavior`, optional `edge_cases_claimed`) that the `verifier` subagent can consume directly.
+
 ## 0.36.0 (2026-06-05)
 
 - **Alibaba DashScope multi-region fallback.** Logging in with a China-region key (`dashscope.aliyuncs.com`) against the default US Virginia endpoint now auto-detects the mismatch and reconfigures for the correct endpoint rather than failing with a misleading "API key is wrong" error.

src/pythinker_code/agents/default/agent.yaml

@@ -47,6 +47,9 @@ agent:
     plan:
       path: ./plan.yaml
       description: "Read-only implementation planning and architecture design."
+    planner:
+      path: ./planner.yaml
+      description: "Read-only recon planner that decomposes tasks into distinct parallel seeds."
     review:
       path: ./review.yaml
       description: "Read-only code review with severity-scored findings."

src/pythinker_code/agents/default/coder.yaml

@@ -31,6 +31,21 @@ agent:
       Bullet list of remaining risks or `None observed.`.
       ### BLOCKERS
       Bullet list of anything that stopped completion, or `None.`.
+
+      Artifact contract: Before finishing, you MUST emit your result as a structured artifact.
+      Wrap it in <coding_artifact> tags on its own line at the very end of your final message:
+
+      <coding_artifact>
+      {
+        "files_changed": ["path/to/file.py"],
+        "test_command": "make test",
+        "expected_behavior": "...",
+        "edge_cases_claimed": ["..."]
+      }
+      </coding_artifact>
+
+      Do not include reasoning, logs, or intermediate output inside the tags — only the JSON fields above.
+      The `edge_cases_claimed` key is optional; omit it if you have no distinct edge cases to claim.
   when_to_use: |
     Use this agent for non-trivial software engineering work that may require reading files, editing code, running commands, and returning a compact but technically complete summary to the parent agent.
   allowed_tools:

src/pythinker_code/agents/default/planner.yaml

@@ -0,0 +1,44 @@
+version: 1
+agent:
+  extend: ./agent.yaml
+  system_prompt_args:
+    ROLE_ADDITIONAL: |
+      You are now running as a subagent. All the `user` messages are sent by the main agent.
+      The main agent cannot see your context, it can only see your last message when you finish.
+
+      You are a Reconnaissance Planner. Your single objective is to analyze the request and
+      break it down into N distinct, non-overlapping task seeds for parallel workers.
+
+      CRITICAL RULES:
+      - Do not solve the problem. Do not write code. Do not fix anything.
+      - Each seed must provide a distinct starting angle (different file, subsystem, or hypothesis)
+        so that parallel workers exploring them will NOT duplicate effort or converge on the same solution.
+      - Aim for 3-5 seeds unless the task is clearly simpler or more complex.
+
+      Final response contract:
+      Your final message must contain ONLY the seeds block below — no preamble, no explanation,
+      no content before or after the tags:
+      <recon_seeds>
+      ["seed description 1", "seed description 2", ...]
+      </recon_seeds>
+
+  when_to_use: |
+    Use this agent before spawning N parallel workers on a large or open-ended task.
+    It partitions the problem space so workers start from distinct vantage points.
+
+  allowed_tools:
+    - "pythinker_code.tools.shell:Shell"
+    - "pythinker_code.tools.file:ReadFile"
+    - "pythinker_code.tools.file:Glob"
+    - "pythinker_code.tools.file:Grep"
+    - "pythinker_code.tools.file:SmartSearch"
+  exclude_tools:
+    - "pythinker_code.tools.agent:Agent"
+    - "pythinker_code.tools.ask_user:AskUserQuestion"
+    - "pythinker_code.tools.plan:ExitPlanMode"
+    - "pythinker_code.tools.plan.enter:EnterPlanMode"
+    - "pythinker_code.tools.file:WriteFile"
+    - "pythinker_code.tools.file:StrReplaceFile"
+    - "pythinker_code.tools.web:SearchWeb"
+    - "pythinker_code.tools.web:FetchURL"
+  subagents:

src/pythinker_code/agents/default/verifier.yaml

@@ -33,6 +33,17 @@ agent:
       Bullet list of likely causes or follow-up fixes, or `None observed.`.
       ### BLOCKERS
       Bullet list of missing dependencies, unavailable commands, or `None.`.
+
+      Artifact receipt: Your input contains a <coding_artifact> block. You will receive ONLY the
+      structured fields from the coder — no conversation history, logs, or confidence scores.
+
+      Your job:
+      1. Run artifact.test_command independently via Shell.
+      2. Check that observed behavior matches artifact.expected_behavior.
+      3. Actively try to break each claim in artifact.edge_cases_claimed.
+      4. Report PASS / FAIL / FLAKY based solely on what you observe — not what the coder claimed.
+
+      Do not ask why the coder made their choices. You have only the artifact.
   when_to_use: |
     Use this agent when the parent needs tests, lint, type checks, builds, or other validation gates run and reported without applying fixes.
   allowed_tools:

src/pythinker_code/tools/file/read.py

@@ -12,6 +12,7 @@
 from pythinker_code.utils.logging import logger
 from pythinker_code.utils.path import is_within_workspace, list_directory
 from pythinker_code.utils.sensitive import is_sensitive_file
+from pythinker_code.utils.trust import UntrustedData
 
 MAX_LINES = 1000
 MAX_LINE_LENGTH = 2000
@@ -128,7 +129,7 @@ async def __call__(self, params: Params) -> ToolReturnValue:
                 )
             if await p.is_dir():
                 return ToolOk(
-                    output=await list_directory(p),
+                    output=UntrustedData(await list_directory(p)).render_for_prompt(),
                     message=(
                         f"Directory listing for `{params.path}`. "
                         "Use ReadFile on a file path to read file contents."
@@ -234,7 +235,7 @@ async def _read_forward(self, p: HostPath, params: Params) -> ToolReturnValue:
         if truncated_line_numbers:
             message += f" Lines {truncated_line_numbers} were truncated."
         return ToolOk(
-            output="".join(lines_with_no),
+            output=UntrustedData("".join(lines_with_no)).render_for_prompt(),
             message=message,
         )
 
@@ -308,6 +309,6 @@ async def _read_tail(self, p: HostPath, params: Params) -> ToolReturnValue:
         if truncated_line_numbers:
             message += f" Lines {truncated_line_numbers} were truncated."
         return ToolOk(
-            output="".join(lines_with_no),
+            output=UntrustedData("".join(lines_with_no)).render_for_prompt(),
             message=message,
         )

src/pythinker_code/tools/web/fetch.py

@@ -18,6 +18,7 @@
 from pythinker_code.tools.web._allowlist import host_in_allowlist
 from pythinker_code.utils.aiohttp import new_client_session
 from pythinker_code.utils.logging import logger
+from pythinker_code.utils.trust import UntrustedData
 
 MAX_FETCH_BYTES = 5 * 1024 * 1024
 MAX_FETCH_REDIRECTS = 10  # matches aiohttp's default redirect cap
@@ -204,7 +205,7 @@ async def fetch_with_http_get(
 
                     content_type = response.headers.get(aiohttp.hdrs.CONTENT_TYPE, "").lower()
                     if content_type.startswith(("text/plain", "text/markdown")):
-                        builder.write(resp_text)
+                        builder.write(UntrustedData(resp_text).render_for_prompt())
                         return builder.ok("The returned content is the full content of the page.")
         except TimeoutError:
             logger.warning("FetchURL timed out: url={url}", url=params.url)
@@ -247,7 +248,7 @@ async def fetch_with_http_get(
                 brief="No content extracted",
             )
 
-        builder.write(extracted_text)
+        builder.write(UntrustedData(extracted_text).render_for_prompt())
         return builder.ok("The returned content is the main text content extracted from the page.")
 
     async def _fetch_with_service(self, params: Params) -> ToolReturnValue:
@@ -307,7 +308,7 @@ async def _fetch_with_service(self, params: Params) -> ToolReturnValue:
                         f"Failed to fetch URL via service: response exceeds {max_mb}MB.",
                         brief="Response too large",
                     )
-                builder.write(content)
+                builder.write(UntrustedData(content).render_for_prompt())
                 return builder.ok(
                     "The returned content is the main content extracted from the page."
                 )

src/pythinker_code/utils/artifacts.py

@@ -0,0 +1,97 @@
+"""Boundary-artifact dataclasses exchanged between coder and verifier subagents.
+
+These dataclasses enforce a strict information barrier: verifiers receive ONLY the
+typed fields declared here, never prose or logs from the producer. The shape of
+``to_json`` / ``from_dict`` is a contract surface consumed by subagent prompts.
+
+Two pairs are defined:
+
+* ``CodingArtifact`` / ``VerificationResult`` for the generic coder-to-verifier flow.
+* ``VulnerabilityArtifact`` / ``AuditVerdict`` for the security-specific finder-to-audit
+  flow.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class CodingArtifact:
+    """Producer-side handoff from a coder subagent to a verifier subagent."""
+
+    files_changed: list[str]
+    test_command: str
+    expected_behavior: str
+    edge_cases_claimed: list[str] = field(default_factory=list[str])
+
+    def to_json(self) -> str:
+        return json.dumps(
+            {
+                "files_changed": self.files_changed,
+                "test_command": self.test_command,
+                "expected_behavior": self.expected_behavior,
+                "edge_cases_claimed": self.edge_cases_claimed,
+            },
+            indent=2,
+        )
+
+    @classmethod
+    def from_dict(cls, d: dict[str, Any]) -> CodingArtifact:
+        return cls(
+            files_changed=d["files_changed"],
+            test_command=d["test_command"],
+            expected_behavior=d["expected_behavior"],
+            edge_cases_claimed=d.get("edge_cases_claimed", []),
+        )
+
+
+@dataclass(frozen=True)
+class VerificationResult:
+    """Verifier-side response back to the coder."""
+
+    passed: bool
+    stdout_summary: str
+    stderr_summary: str
+    discovered_gaps: list[str] = field(default_factory=list[str])
+
+
+@dataclass(frozen=True)
+class VulnerabilityArtifact:
+    """Producer-side handoff from a vulnerability finder to the audit verifier."""
+
+    target_file: str
+    vulnerability_type: str
+    reproduction_command: str
+    expected_failure_output: str
+
+    def to_json(self) -> str:
+        return json.dumps(
+            {
+                "target_file": self.target_file,
+                "vulnerability_type": self.vulnerability_type,
+                "reproduction_command": self.reproduction_command,
+                "expected_failure_output": self.expected_failure_output,
+            },
+            indent=2,
+        )
+
+    @classmethod
+    def from_dict(cls, d: dict[str, Any]) -> VulnerabilityArtifact:
+        return cls(
+            target_file=d["target_file"],
+            vulnerability_type=d["vulnerability_type"],
+            reproduction_command=d["reproduction_command"],
+            expected_failure_output=d["expected_failure_output"],
+        )
+
+
+@dataclass(frozen=True)
+class AuditVerdict:
+    """Audit-verifier verdict on a claimed vulnerability."""
+
+    vulnerability_confirmed: bool
+    execution_logs: str
+    false_positive_reasoning: str = ""

src/pythinker_code/utils/trust.py

@@ -0,0 +1,23 @@
+"""Trust-wrapping primitive for external, untrusted data entering LLM prompts."""
+
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class UntrustedData:
+    """Marks a string as originating from an external, untrusted source.
+
+    Call render_for_prompt() before injecting the content into any LLM prompt.
+    The runtime nonce prevents the model from constructing a matching opening
+    tag. The closing-tag escape prevents content from breaking out of the block.
+    """
+
+    raw_content: str
+
+    def render_for_prompt(self) -> str:
+        nonce = uuid.uuid4().hex[:8]
+        safe_content = self.raw_content.replace("</untrusted_data>", "&lt;/untrusted_data&gt;")
+        return f'<untrusted_data id="{nonce}">\n{safe_content}\n</untrusted_data>'