Build a Local Pentest Planning Copilot
Build a local AI workflow for approved pentest planning and coverage reports.
Introduction
30 Second Summary
A security review can scatter risks across several files. Test guidance can drift away from the approval that controls it.
In this project, you will build a local copilot that turns a synthetic SaaS threat model into an approved pentest plan. It pauses for your decision before processing simulated evidence into a traceable coverage report.
What You'll Build
Picture running a local planning workflow that waits for your approval before producing a Markdown coverage report where every risk remains linked to its evidence.
By the end of this project, you'll have:
- A typed scope preview that preserves every synthetic threat ID in a readable JSON artifact.
- A resumable approval workflow where LangGraph pauses before coverage processing. Your decision resumes the same SQLite-backed thread.
- Evidence-based coverage reports that keep partial risks visible. Blocked risks remain visible for manual review.
- Secret Mission: Export every partial, blocked, or manual-required risk into a focused manual test queue.
Are there any prerequisites?
You need Windows 10 22H2 or newer with Python 3.10 or newer. You also need enough free storage for Ollama and its 4.9 GB model.
Use only the included synthetic data because this project never authorizes testing against a real system.
Before We Start
Before any hands-on work, lock in a local, non-executing pentest-planning workflow that uses synthetic data only. Your approval remains authoritative before simulated evidence processing.
Set Up the Local Agent Stack
A dependable local copilot needs the same Python packages every time. It also needs a local model that responds before you start writing agent code.
You will use PowerShell to create an isolated Python environment. Ollama will run the language model on your computer using only synthetic project inputs.
In this step, get ready to:
- Create a project folder with an isolated Python environment.
- Install Ollama with the local model.
- Install the pinned Python packages.
Prepare the Python environment
A virtual environment gives this project its own package installation. This prevents packages from other Python projects from changing how your copilot behaves.
- Press the Windows key to open search.
- Type PowerShell into search.
- Press Enter to open PowerShell.
- Check your installed Python version by running:
python --version
What does this command check?
The --version option prints the Python version used by the python command. This project requires Python 3.10 or newer.
✔️ I see version 3.10 or higher
Your Python version supports every pinned package in this project.
- Continue to the project-folder instructions below.
ⓧ I see an older version
Your current Python version cannot run the pinned agent packages. Install a current supported release before creating the environment.
- Open the official Python releases for Windows in your browser.
- Download a current supported Windows installer.
- Run the downloaded installer.
- Enable the option that adds Python to your PATH when the installer presents it.
- Close PowerShell after installation completes.
- Open a fresh PowerShell window through Windows search.
- Check the Python version again by running:
python --version
What should I see?
The command should now report Python 3.10 or newer. The fresh PowerShell window loads the updated PATH.
ⓧ Command not found
PowerShell cannot currently find a Python installation. The official Windows installer provides Python with the tools needed for this project.
- Open the official Python releases for Windows in your browser.
- Download a current supported Windows installer.
- Run the downloaded installer.
- Enable the option that adds Python to your PATH when the installer presents it.
- Close PowerShell after installation completes.
- Open a fresh PowerShell window through Windows search.
- Check the Python version by running:
python --version
What should I see?
PowerShell should print Python 3.10 or newer. That confirms the installer made the interpreter available to your terminal.
- Move to your Desktop by running this command:
Set-Location "$env:USERPROFILE\Desktop"
What does this command do?
PowerShell changes its current location to the Desktop folder for your Windows account. The project folder you create next will be easy to find.
- Create the pentest-copilot folder by running:
New-Item -ItemType Directory -Path ".\pentest-copilot"
What does this command create?
PowerShell creates the pentest-copilot folder on your Desktop. You should see the new directory listed in the command output.
- Enter the pentest-copilot folder by running:
Set-Location ".\pentest-copilot" -PassThru
What should I see?
The -PassThru option prints the new location. The path should end with Desktop\pentest-copilot.
- Create the .venv virtual environment by running:
python -m venv .venv
What does this command create?
Python creates a separate interpreter with its own package directory inside .venv. Later commands call that interpreter directly, so activation is unnecessary.
- Confirm that the virtual environment exists by running:
Get-ChildItem -Force
What should I see?
The folder listing should include .venv. That directory proves the project has its own Python environment.
That is the first foundation in place. Your copilot now has an isolated Python workspace.
- Open the current project folder in Visual Studio Code by running:
code .
What does this command open?
The period represents the current pentest-copilot folder. Visual Studio Code opens that folder as your workspace.
Project folder not opening?
If PowerShell cannot launch Visual Studio Code, close PowerShell first. Open Visual Studio Code through Windows search instead.
Use the editor's folder-opening option to select Desktop\pentest-copilot. This reaches the same workspace without the command-line shortcut.
Help me open my pentest-copilot folder in Visual Studio Code on Windows.
Install Ollama and the local model
Ollama provides the local inference service that your Python workflow will call. The model stays on your computer, so this project does not need a hosted AI API.
- Return to the PowerShell window from earlier.
- Check whether Ollama is already installed by running:
ollama -v
What does this command check?
The command asks the installed Ollama application to print its version. Use the matching tab below for the result you see.
✔️ I see version 0.40.15 or higher
Ollama 0.40.15 or newer is ready to run the local model.
- Continue to the model download below.
ⓧ I see an older version
Your Ollama installation needs an update before it runs the pinned local stack.
- Open the official Ollama Windows guide in your browser.
- Download OllamaSetup.exe.
- Run the downloaded installer.
- Close PowerShell after installation completes.
- Open a fresh PowerShell window through Windows search.
- Check the updated Ollama version by running:
ollama -v
What should I see?
The command should report Ollama 0.40.15 or newer. The refreshed terminal can now find the updated application.
ⓧ Command not found
Ollama is not available to PowerShell yet. The official Windows installer sets up the native application with its local service.
- Open the official Ollama Windows guide in your browser.
- Download OllamaSetup.exe.
- Run the downloaded installer.
- Close PowerShell after installation completes.
- Open a fresh PowerShell window through Windows search.
- Check the Ollama version by running:
ollama -v
What should I see?
The command should print an Ollama version. A version of 0.40.15 or newer is ready for this project.
The llama3.1:8b model is a 4.9 GB download. Expect this command to take several minutes while Ollama displays download progress.
- Download the model and open its interactive prompt by running:
ollama run llama3.1:8b
What does this command do?
Ollama downloads the exact llama3.1:8b model when it is missing. It then opens a local chat prompt after the download completes.
- Send the readiness prompt by entering:
Reply with READY only
What does this prompt prove?
The prompt asks for one unmistakable response from the local model. A reply containing READY proves the model can process a request on your computer.
You should see a local response containing READY. Your model is now available without contacting a target system.
- Leave the interactive model session by entering:
/bye
What does this command do?
The /bye command closes the interactive chat. PowerShell returns to the project folder prompt.
Model not responding?
Confirm that the Ollama application is running on Windows. Restart it through Windows search if the local service stopped.
Confirm that the model command uses the exact llama3.1:8b tag. A different tag refers to a different model artifact.
Help me diagnose why llama3.1:8b is not downloading or responding through Ollama on Windows.
Install the pinned Python packages
Pinned dependencies keep every learner on the same package interfaces. LangGraph controls the later workflow stages while deterministic Python code protects authorization decisions.
- Switch back to the pentest-copilot workspace in Visual Studio Code.
- Select the new-file control in the Explorer sidebar.
- Enter requirements.txt as the file name.
- Add the pinned package list to requirements.txt by pasting:
langgraph==1.2.20
langgraph-checkpoint-sqlite==3.1.3
ollama==0.6.3
pydantic==2.14.3
What do these packages provide?
- The langgraph==1.2.20 package provides the deterministic graph orchestration used later.
- The langgraph-checkpoint-sqlite==3.1.3 package adds persistent checkpoints through SQLite.
- The ollama==0.6.3 package provides the official Ollama Python client.
- The pydantic==2.14.3 package provides typed validation through Pydantic models.
- Save requirements.txt by pressing Ctrl+S.
✔️ My requirements file matches
Your dependency file is saved with all four pinned packages.
ⓧ I'd like to double check the full code
Compare your saved requirements.txt file with this complete version.
langgraph==1.2.20
langgraph-checkpoint-sqlite==3.1.3
ollama==0.6.3
pydantic==2.14.3
Package installation can take a few minutes while pip resolves each pinned dependency. Keep the PowerShell window open until the prompt returns.
- Return to PowerShell.
- Install the pinned packages into .venv by running:
.\.venv\Scripts\python.exe -m pip install -r requirements.txt
What does this command do?
The command runs pip through the Python interpreter inside .venv. The -r option installs every pinned entry from requirements.txt.
The PowerShell prompt should return without an installation error. Your isolated environment now contains the four required packages.
Package installation failing?
Confirm that PowerShell is still inside the pentest-copilot folder. The command depends on finding both .venv and requirements.txt there.
Confirm that your Python version is 3.10 or newer. Older interpreters cannot install every pinned package.
Help me diagnose why the pinned requirements are not installing into my Windows virtual environment.
Before you run the import check, do you expect all three top-level packages to load from the new virtual environment?
- Verify the installed Python packages by running:
.\.venv\Scripts\python.exe -c "import langgraph, ollama, pydantic; print('agent stack ready')"
What does this check prove?
Python imports LangGraph, the Ollama client, and Pydantic from the virtual environment. The success message prints only after every import completes.
You should see agent stack ready in PowerShell. That confirms the Python side of your local stack is ready.
Import check not passing?
Rerun the package installation from the pentest-copilot folder. Make sure the install command finishes before running the import check again.
Confirm that both commands begin with .\.venv\Scripts\python.exe. This keeps installation and verification inside the same environment.
Help me find which pinned package is causing my local agent stack import check to fail.
One final check confirms that the Python stack and local inference service are both available. Before you run it, what response do you expect from the model?
- Open the local model prompt again by running:
ollama run llama3.1:8b
What happens this time?
Ollama reuses the model already stored on your computer. The interactive prompt should open without downloading the 4.9 GB artifact again.
- Send the final readiness check by entering:
Reply with READY only
What should I see?
The terminal should print a local model response containing READY. This is the end-of-step proof that Ollama can serve the pinned model.
You now have both success signals. PowerShell prints agent stack ready for the packages and READY for the local model.
- Return to the PowerShell prompt by entering:
/bye
What happens next?
The model session closes while Ollama remains installed on your computer. Your project stays limited to synthetic documents and simulated evidence.
Your reproducible local stack is ready. Next, you will turn the bundled synthetic threat data into a typed scope preview.
Generate a Typed Scope Preview
Your local model can respond inside the verified Python environment. The next challenge is keeping its scope output consistent enough to validate.
Free-form scope prose makes identifiers difficult to trace. This step uses Pydantic to constrain the response as typed JSON built from fixed threat IDs.
In this step, get ready to:
- Add the synthetic security inputs under data.
- Define the typed scope models and local inference workflow.
- Generate a scope preview with one item for each supplied threat.
Create the synthetic input set
The copilot needs fixed inputs before it can draft a trustworthy scope. These fixtures keep every activity synthetic while preserving the relationships between threats, risks, references, and simulated evidence.
The reference catalog bundles selected entries from the OWASP Web Security Testing Guide and MITRE ATT&CK®. The mappings in the risk table belong to this educational project.
- Use the file controls in the existing Visual Studio Code sidebar to create a folder named data inside the project folder.
- Create threat-model.md inside data.
- Add the synthetic product boundaries by pasting this content into data/threat-model.md:
# Acme Admin Portal Threat Model
Product: A synthetic multi-tenant SaaS administration portal.
Components:
- Internet-facing login page
- Tenant administration interface
- REST API used by the interface
Trust boundaries:
- The public internet to the login service
- An authenticated tenant to tenant-owned objects
- The public API edge to backend application services
Threats:
- THR-001: Credential stuffing against the login page may compromise a tenant administrator.
- THR-002: Changing a tenant identifier may expose another tenant's objects.
- THR-003: Crafted input sent to the public API may exploit an application weakness.
Safety boundary:
This project uses documents and simulated evidence only. It does not send traffic to a target.
What Does This Input Control?
- The component list gives the scope generator concrete assets to describe.
- The three threat IDs become the fixed identifiers that the generated scope must preserve.
- The safety boundary prevents the planning workflow from contacting a target.
- Save data/threat-model.md.
You should see three threat entries followed by a synthetic-only safety boundary.
Missing a Threat Entry?
Check that THR-001, THR-002, and THR-003 each appear once in data/threat-model.md.
Ask for help checking the fixture if the identifiers do not match. Help me compare data/threat-model.md with the required synthetic threat model.
- Create security-requirements.md inside data.
- Add the authorization requirements by pasting this content into data/security-requirements.md:
# Security Requirements
- REQ-001: Authentication controls must resist repeated credential attacks and preserve evidence of enforcement.
- REQ-002: Every tenant object request must enforce ownership and role authorization on the server.
- REQ-003: Internet-facing API input must be validated before it reaches privileged application behavior.
- REQ-004: Testing requires an approved scope and must use synthetic or explicitly authorized evidence.
Why These Requirements Matter
These requirements define the controls that the scope should address. REQ-004 keeps authorization and synthetic evidence inside the generated boundaries.
- Save data/security-requirements.md.
You should see four requirements covering authentication, authorization, input validation, and approval.
Requirements Look Incomplete?
Confirm that the requirement IDs run from REQ-001 through REQ-004.
Use this help prompt if an entry is missing. Help me find a missing requirement in data/security-requirements.md.
- Create risk-table.json inside data.
- Add the deterministic threat-to-risk mappings by pasting this content into data/risk-table.json:
[
{
"risk_id": "RISK-001",
"threat_id": "THR-001",
"title": "Credential attack resistance",
"asset": "Login service",
"source_refs": ["WSTG-ATHN-03", "T1110"]
},
{
"risk_id": "RISK-002",
"threat_id": "THR-002",
"title": "Cross-tenant authorization",
"asset": "Tenant object API",
"source_refs": ["WSTG-ATHZ-02"]
},
{
"risk_id": "RISK-003",
"threat_id": "THR-003",
"title": "Public API exploitation path",
"asset": "Internet-facing REST API",
"source_refs": ["T1190"]
}
]
What Does the Risk Table Preserve?
- Each risk_id identifies one security concern.
- Each threat_id links that concern back to the supplied threat model.
- Each source_refs list points to bundled reference material.
- Save data/risk-table.json.
You should see three risk objects with one matching threat ID in each object.
Seeing JSON Syntax Markers?
Check every quoted key for a matching closing quote. Check for a comma between adjacent objects.
Use this prompt to locate the invalid character. Help me find the JSON syntax issue in data/risk-table.json.
- Create reference-catalog.json inside data.
- Add the bundled source snapshot by pasting this content into data/reference-catalog.json:
[
{
"id": "WSTG-ATHN-03",
"title": "Weak Lock Out Mechanism",
"focus": "Evaluate the account lockout mechanism's ability to mitigate brute force password guessing.",
"url": "https://wstg.owasp.org/latest/4-Web_Application_Security_Testing/04-Authentication/03-Weak_Lock_Out_Mechanism/"
},
{
"id": "WSTG-ATHZ-02",
"title": "Bypassing Authorization Schema",
"focus": "Assess if unauthenticated, horizontal, or vertical access is possible.",
"url": "https://wstg.owasp.org/latest/4-Web_Application_Security_Testing/05-Authorization/02-Bypassing_Authorization_Schema/"
},
{
"id": "T1110",
"title": "Brute Force",
"focus": "Adversaries may use brute force techniques to gain access to accounts when passwords are unknown or when password hashes are obtained.",
"url": "https://attack.mitre.org/techniques/T1110/"
},
{
"id": "T1190",
"title": "Exploit Public-Facing Application",
"focus": "Adversaries may attempt to exploit a weakness in an Internet-facing host or system to initially access a network.",
"url": "https://attack.mitre.org/techniques/T1190/"
}
]
How the Catalog Grounds the Model
The catalog gives the local model a fixed title and focus statement for every source reference. Later stages can summarize these entries without inventing new source mappings.
- Save data/reference-catalog.json.
You should see four catalog entries matching every source reference used by the risk table.
Source References Do Not Match?
Compare the id values with the source_refs values in data/risk-table.json.
Ask for a reference comparison if one value differs. Help me compare the risk table references with the reference catalog IDs.
- Create automation-results.json inside data.
- Add the simulated evidence fixture by pasting this content into data/automation-results.json:
[
{
"risk_id": "RISK-001",
"source": "simulated-auth-check",
"status": "partial",
"evidence": "The fixture records repeated failed logins, but distributed-source behavior and recovery controls were not assessed."
},
{
"risk_id": "RISK-002",
"source": "simulated-tenant-id-check",
"status": "automated-covered",
"evidence": "The fixture records a denied cross-tenant object request using a second tenant identifier."
},
{
"risk_id": "RISK-003",
"source": "simulated-input-check",
"status": "blocked",
"evidence": "The fixture contains no approved non-production endpoint or safe execution evidence."
}
]
Why the Evidence Is Simulated
This fixture represents evidence that a future execution adapter could supply. It lets the reporting workflow handle covered, partial, and blocked states without scanning any system.
- Save data/automation-results.json.
You should see one simulated result for each risk ID.
Evidence Fixture Looks Invalid?
Confirm that each object contains risk_id, source, status, and evidence.
Use this prompt to check the three objects. Help me validate data/automation-results.json without changing its evidence.
Define the typed preview pipeline
A schema gives the model a fixed output shape. Deterministic checks then confirm that its response contains the same threat IDs as the supplied risk table.
- Create schemas.py inside the open project folder.
- Add the coverage status and scope models by pasting this first chunk into schemas.py:
from typing import Literal
from pydantic import BaseModel
CoverageStatus = Literal[
"automated-covered",
"partial",
"manual-required",
"blocked",
"out-of-scope",
]
class ScopeItem(BaseModel):
scope_id: str
threat_id: str
asset: str
test_objective: str
boundaries: list[str]
authorization_required: bool
class ScopeDraft(BaseModel):
engagement_id: str
product: str
assumptions: list[str]
exclusions: list[str]
items: list[ScopeItem]
What Do the Scope Models Enforce?
- CoverageStatus limits later coverage objects to known status values.
- ScopeItem requires every generated item to include a threat ID and authorization boundary.
- ScopeDraft wraps the complete preview under one engagement ID.
- Save schemas.py.
You should see ScopeItem and ScopeDraft listed beneath the allowed coverage statuses.
Seeing a Schema Syntax Marker?
Check the closing bracket after CoverageStatus. Check the indentation of every field beneath each class.
Use this prompt to review the first schema chunk. Help me find the syntax issue in the scope models.
- Add the enrichment models directly below ScopeDraft by pasting this chunk:
class EnrichmentNarrative(BaseModel):
risk_id: str
source_summary: str
risk_rationale: str
class EnrichmentNarratives(BaseModel):
items: list[EnrichmentNarrative]
class EnrichedRisk(BaseModel):
risk_id: str
threat_id: str
source_refs: list[str]
source_summary: str
risk_rationale: str
class EnrichmentBundle(BaseModel):
items: list[EnrichedRisk]
How Enrichment Stays Traceable
The narrative model holds generated explanation. EnrichedRisk adds deterministic threat IDs and source references around that explanation.
- Save schemas.py.
You should now see four enrichment classes beneath the scope models.
Enrichment Classes Nested Accidentally?
Align each class declaration with the left edge of class ScopeDraft.
Ask for indentation help if a class appears inside another class. Help me fix the enrichment model indentation.
- Add the guidance and approval models below EnrichmentBundle by pasting this chunk:
class GuidanceNarrative(BaseModel):
risk_id: str
test_title: str
safe_steps: list[str]
expected_evidence: str
manual_notes: str
class GuidanceNarratives(BaseModel):
items: list[GuidanceNarrative]
class TestGuidance(BaseModel):
risk_id: str
threat_id: str
source_refs: list[str]
test_title: str
safe_steps: list[str]
expected_evidence: str
manual_notes: str
class GuidanceBundle(BaseModel):
items: list[TestGuidance]
class ApprovalDecision(BaseModel):
approved: bool
note: str | None = None
What These Models Separate
The guidance narrative contains safe engineer-facing text. TestGuidance keeps the deterministic risk, threat, and reference identifiers attached.
ApprovalDecision gives the later interrupt a typed human decision.
- Save schemas.py.
You should see the guidance bundle followed by a separate approval decision model.
Approval Note Type Underlined?
Confirm that the existing virtual environment uses Python 3.10 or newer. The union type on note depends on that version.
Use this prompt if the interpreter selection looks wrong. Help me check why the approval model type is underlined.
- Add the automation and coverage models below ApprovalDecision by pasting this final chunk:
class AutomationResult(BaseModel):
risk_id: str
source: str
status: Literal["automated-covered", "partial", "blocked"]
evidence: str
class CoverageItem(BaseModel):
risk_id: str
threat_id: str
status: CoverageStatus
evidence_source: str
evidence: str
source_refs: list[str]
manual_required: bool
class CoverageReport(BaseModel):
engagement_id: str
approved: bool
approval_note: str
summary: str
items: list[CoverageItem]
How Coverage Remains Evidence-Based
AutomationResult accepts only the statuses present in the simulated fixture. CoverageItem keeps the evidence and traceability fields together.
CoverageReport records whether approval was granted before evidence processing.
- Save schemas.py.
You should now see the complete model set ending with CoverageReport.
Coverage Models Showing Undefined Names?
Confirm that CoverageStatus remains near the top of schemas.py. Confirm that Literal is still imported.
Ask for a dependency check if either name is missing. Help me trace the undefined name in the coverage models.
✔️ Awesome, I've got everything!
Great. Double check that schemas.py is saved and ends with CoverageReport.
ⓧ I'd like to double check the full code
Compare your complete schemas.py file with this reference:
from typing import Literal
from pydantic import BaseModel
CoverageStatus = Literal[
"automated-covered",
"partial",
"manual-required",
"blocked",
"out-of-scope",
]
class ScopeItem(BaseModel):
scope_id: str
threat_id: str
asset: str
test_objective: str
boundaries: list[str]
authorization_required: bool
class ScopeDraft(BaseModel):
engagement_id: str
product: str
assumptions: list[str]
exclusions: list[str]
items: list[ScopeItem]
class EnrichmentNarrative(BaseModel):
risk_id: str
source_summary: str
risk_rationale: str
class EnrichmentNarratives(BaseModel):
items: list[EnrichmentNarrative]
class EnrichedRisk(BaseModel):
risk_id: str
threat_id: str
source_refs: list[str]
source_summary: str
risk_rationale: str
class EnrichmentBundle(BaseModel):
items: list[EnrichedRisk]
class GuidanceNarrative(BaseModel):
risk_id: str
test_title: str
safe_steps: list[str]
expected_evidence: str
manual_notes: str
class GuidanceNarratives(BaseModel):
items: list[GuidanceNarrative]
class TestGuidance(BaseModel):
risk_id: str
threat_id: str
source_refs: list[str]
test_title: str
safe_steps: list[str]
expected_evidence: str
manual_notes: str
class GuidanceBundle(BaseModel):
items: list[TestGuidance]
class ApprovalDecision(BaseModel):
approved: bool
note: str | None = None
class AutomationResult(BaseModel):
risk_id: str
source: str
status: Literal["automated-covered", "partial", "blocked"]
evidence: str
class CoverageItem(BaseModel):
risk_id: str
threat_id: str
status: CoverageStatus
evidence_source: str
evidence: str
source_refs: list[str]
manual_required: bool
class CoverageReport(BaseModel):
engagement_id: str
approved: bool
approval_note: str
summary: str
items: list[CoverageItem]
The typed models define the output contract. The next file connects that contract to the local Ollama model and rejects any response that changes the supplied threat set.
- Create workflow.py inside the open project folder.
- Add the imports and shared paths by pasting this first chunk into workflow.py:
import json
from pathlib import Path
from typing import TypedDict
from ollama import chat
from pydantic import BaseModel
from schemas import (
ScopeDraft,
)
ROOT = Path(__file__).resolve().parent
DATA_DIR = ROOT / "data"
OUTPUT_DIR = ROOT / "outputs"
MODEL = "llama3.1:8b"
What Do These Paths Represent?
- ROOT anchors every path to the folder containing workflow.py.
- DATA_DIR points to the synthetic input folder.
- OUTPUT_DIR points to the generated engagement artifacts.
- MODEL pins local inference to llama3.1:8b.
- Save workflow.py.
You should see the three project paths followed by the pinned local model tag.
Imports Underlined in workflow.py?
Confirm that schemas.py is saved in the same project folder. Confirm that the existing virtual environment remains selected.
Use this prompt to trace the import location. Help me diagnose the workflow.py import path.
- Add the workflow state and file helpers below MODEL by pasting this chunk:
class WorkflowState(TypedDict, total=False):
engagement_id: str
scope: dict
enriched_risks: dict
guidance: dict
approved: bool
approval_note: str
coverage: dict
report_path: str
def load_json(path: Path):
return json.loads(path.read_text(encoding="utf-8"))
def write_model(path: Path, model: BaseModel) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(model.model_dump_json(indent=2), encoding="utf-8")
def engagement_dir(engagement_id: str) -> Path:
path = OUTPUT_DIR / engagement_id
path.mkdir(parents=True, exist_ok=True)
return path
What Do the Helpers Do?
- WorkflowState names the artifacts that later graph stages can carry.
- load_json() reads a bundled JSON fixture.
- write_model() serializes a validated model into an output file.
- engagement_dir() gives each engagement its own output folder.
- Save workflow.py.
You should see three helper functions beneath the workflow state definition.
Helper Indentation Looks Wrong?
Align each def declaration with the left edge of class WorkflowState.
Use this prompt to inspect the function boundaries. Help me fix the helper indentation in workflow.py.
- Add the structured local model call below engagement_dir() by pasting this chunk:
def call_structured(schema: type[BaseModel], system_prompt: str, payload: dict) -> BaseModel:
schema_dict = schema.model_json_schema()
response = chat(
model=MODEL,
messages=[
{"role": "system", "content": system_prompt},
{
"role": "user",
"content": (
"Use only this input:\n"
+ json.dumps(payload, indent=2)
+ "\nReturn data matching this JSON schema:\n"
+ json.dumps(schema_dict, indent=2)
),
},
],
format=schema_dict,
options={"temperature": 0},
)
return schema.model_validate_json(response.message.content)
How Structured Local Inference Works
- model_json_schema() turns the selected Pydantic model into a JSON schema.
- chat() sends the supplied payload and schema to the local model.
- format requests a response matching the schema.
- model_validate_json() rejects a response that does not match the selected model.
- Save workflow.py.
You should see the validated return statement at the end of call_structured().
Structured Call Showing Mismatched Brackets?
Check the closing brackets for messages. Check that the closing parenthesis for chat() appears before the return statement.
Use this prompt to pair the delimiters. Help me find the unmatched delimiter in call_structured().
- Start draft_scope() below call_structured() by pasting this first part:
def draft_scope(state: WorkflowState) -> dict:
engagement_id = state["engagement_id"]
risks = load_json(DATA_DIR / "risk-table.json")
payload = {
"engagement_id": engagement_id,
"threat_model": (DATA_DIR / "threat-model.md").read_text(encoding="utf-8"),
"security_requirements": (DATA_DIR / "security-requirements.md").read_text(
encoding="utf-8"
),
"risk_table": risks,
}
What Does the Scope Payload Contain?
The payload combines the engagement ID with the threat model, security requirements, and deterministic risk table. The model receives no target address or active testing input.
- Save workflow.py.
You should see the payload dictionary close directly beneath the risk table entry.
Payload Paths Do Not Match?
Compare the three filenames with the files under data. Keep every hyphen and file extension unchanged.
Ask for a path comparison if one file is unresolved. Help me compare the draft_scope data paths with my files.
- Complete draft_scope() by adding this code directly below the payload:
scope = call_structured(
ScopeDraft,
(
"You are a defensive penetration-test scope planner. Create exactly one scope "
"item per threat_id. Copy all identifiers exactly. Keep every activity synthetic "
"and non-executing. State that authorization is required. Do not add targets, "
"payloads, exploit commands, or facts outside the supplied input."
),
payload,
)
scope = ScopeDraft.model_validate(
{**scope.model_dump(), "engagement_id": engagement_id}
)
expected_threats = {risk["threat_id"] for risk in risks}
actual_threats = {item.threat_id for item in scope.items}
if actual_threats != expected_threats or len(scope.items) != len(risks):
raise ValueError("Scope output did not preserve the supplied threat IDs exactly.")
write_model(engagement_dir(engagement_id) / "scope.json", scope)
return {"scope": scope.model_dump()}
Where the Deterministic Guardrail Lives
- The system prompt limits the model to defensive and synthetic planning.
- expected_threats comes directly from the deterministic risk table.
- actual_threats comes from the validated model response.
- The comparison stops the workflow if an identifier is missing or changed.
- write_model() saves the accepted preview as scope.json.
- Save workflow.py.
You should see draft_scope() end with a dictionary containing the validated scope.
draft_scope Looks Split in Two?
Keep the second chunk indented by four spaces so it remains inside draft_scope().
Use this prompt to review the function boundary. Help me check the draft_scope indentation.
✔️ Awesome, I've got everything!
Good work. workflow.py now loads the synthetic inputs, requests a typed local response, and guards the supplied threat IDs.
ⓧ I'd like to double check the full code
Compare your complete workflow.py file with this scope-preview reference:
import json
from pathlib import Path
from typing import TypedDict
from ollama import chat
from pydantic import BaseModel
from schemas import (
ScopeDraft,
)
ROOT = Path(__file__).resolve().parent
DATA_DIR = ROOT / "data"
OUTPUT_DIR = ROOT / "outputs"
MODEL = "llama3.1:8b"
class WorkflowState(TypedDict, total=False):
engagement_id: str
scope: dict
enriched_risks: dict
guidance: dict
approved: bool
approval_note: str
coverage: dict
report_path: str
def load_json(path: Path):
return json.loads(path.read_text(encoding="utf-8"))
def write_model(path: Path, model: BaseModel) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(model.model_dump_json(indent=2), encoding="utf-8")
def engagement_dir(engagement_id: str) -> Path:
path = OUTPUT_DIR / engagement_id
path.mkdir(parents=True, exist_ok=True)
return path
def call_structured(schema: type[BaseModel], system_prompt: str, payload: dict) -> BaseModel:
schema_dict = schema.model_json_schema()
response = chat(
model=MODEL,
messages=[
{"role": "system", "content": system_prompt},
{
"role": "user",
"content": (
"Use only this input:\n"
+ json.dumps(payload, indent=2)
+ "\nReturn data matching this JSON schema:\n"
+ json.dumps(schema_dict, indent=2)
),
},
],
format=schema_dict,
options={"temperature": 0},
)
return schema.model_validate_json(response.message.content)
def draft_scope(state: WorkflowState) -> dict:
engagement_id = state["engagement_id"]
risks = load_json(DATA_DIR / "risk-table.json")
payload = {
"engagement_id": engagement_id,
"threat_model": (DATA_DIR / "threat-model.md").read_text(encoding="utf-8"),
"security_requirements": (DATA_DIR / "security-requirements.md").read_text(
encoding="utf-8"
),
"risk_table": risks,
}
scope = call_structured(
ScopeDraft,
(
"You are a defensive penetration-test scope planner. Create exactly one scope "
"item per threat_id. Copy all identifiers exactly. Keep every activity synthetic "
"and non-executing. State that authorization is required. Do not add targets, "
"payloads, exploit commands, or facts outside the supplied input."
),
payload,
)
scope = ScopeDraft.model_validate(
{**scope.model_dump(), "engagement_id": engagement_id}
)
expected_threats = {risk["threat_id"] for risk in risks}
actual_threats = {item.threat_id for item in scope.items}
if actual_threats != expected_threats or len(scope.items) != len(risks):
raise ValueError("Scope output did not preserve the supplied threat IDs exactly.")
write_model(engagement_dir(engagement_id) / "scope.json", scope)
return {"scope": scope.model_dump()}
Generate the preview
The command-line entry point turns an engagement ID into a validated scope artifact. It also enables strict checkpoint serialization before later steps add persistence.
- Create app.py inside the open project folder.
- Add the environment setup and imports by pasting this first chunk into app.py:
import os
import sys
from pathlib import Path
os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")
from schemas import ScopeDraft
from workflow import ROOT, draft_scope
Why Set the Environment First?
The environment setting appears before project imports so later SQLite checkpoint code inherits strict serialization behavior. The preview imports only the scope model and scope-drafting function it needs.
- Save app.py.
You should see the environment setting between the standard library imports and the project imports.
Project Imports Underlined in app.py?
Confirm that schemas.py and workflow.py are saved beside app.py.
Use this prompt to compare the file locations. Help me diagnose the local imports in app.py.
- Add the configuration helper and preview command below the imports by pasting this chunk:
def config_for(engagement_id: str) -> dict:
return {"configurable": {"thread_id": engagement_id}}
def preview(engagement_id: str) -> None:
result = draft_scope({"engagement_id": engagement_id})
scope = ScopeDraft.model_validate(result["scope"])
print(f"SCOPE PREVIEW: {len(scope.items)} items")
print(ROOT / "outputs" / engagement_id / "scope.json")
What Does preview() Confirm?
preview() invokes the typed scope workflow for the supplied engagement ID. It validates the returned object again before printing the item count and output path.
config_for() prepares the matching thread configuration that the persisted workflow uses in the next step.
- Save app.py.
You should see preview() end with two output lines.
Preview Function Nested Under config_for?
Align def preview with def config_for at the left edge of the file.
Ask for an indentation check if the functions appear nested. Help me check the function boundaries in app.py.
- Add the command dispatcher below preview() by pasting this final chunk:
def usage() -> None:
script = Path(sys.argv[0]).name
print(f"Usage: {script} preview|plan|approve|reject|audit <engagement-id>")
def main() -> None:
if len(sys.argv) != 3:
usage()
raise SystemExit(2)
command, engagement_id = sys.argv[1], sys.argv[2]
if command == "preview":
preview(engagement_id)
else:
usage()
raise SystemExit(2)
if __name__ == "__main__":
main()
How the Command Reaches the Workflow
- usage() prints the available command shape when the arguments are invalid.
- main() separates the command from the engagement ID.
- The preview branch passes the engagement ID to preview().
- The final condition runs main() when PowerShell starts the file.
- Save app.py.
You should see main() as the final function in the file.
Command Dispatcher Showing an Indentation Issue?
Confirm that if __name__ == "__main__": begins at the left edge. Keep only its call to main() indented.
Use this prompt to inspect the final block. Help me fix the app.py command dispatcher indentation.
✔️ Awesome, I've got everything!
Your command-line entry point is ready. Double check that app.py is saved before generating the preview.
ⓧ I'd like to double check the full code
Compare your complete app.py file with this reference:
import os
import sys
from pathlib import Path
os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")
from schemas import ScopeDraft
from workflow import ROOT, draft_scope
def config_for(engagement_id: str) -> dict:
return {"configurable": {"thread_id": engagement_id}}
def preview(engagement_id: str) -> None:
result = draft_scope({"engagement_id": engagement_id})
scope = ScopeDraft.model_validate(result["scope"])
print(f"SCOPE PREVIEW: {len(scope.items)} items")
print(ROOT / "outputs" / engagement_id / "scope.json")
def usage() -> None:
script = Path(sys.argv[0]).name
print(f"Usage: {script} preview|plan|approve|reject|audit <engagement-id>")
def main() -> None:
if len(sys.argv) != 3:
usage()
raise SystemExit(2)
command, engagement_id = sys.argv[1], sys.argv[2]
if command == "preview":
preview(engagement_id)
else:
usage()
raise SystemExit(2)
if __name__ == "__main__":
main()
Before you run the preview, do you think the local model will preserve all three supplied threat IDs?
- Generate the typed scope preview from the existing PowerShell terminal by running this command:
.\.venv\Scripts\python.exe app.py preview demo-preview
What Should You See?
The local model processes the bundled documents before the command returns. You should see SCOPE PREVIEW: 3 items followed by the path to outputs/demo-preview/scope.json.
That count proves the generated preview preserved one item for each supplied threat.
Preview Command Did Not Complete?
- Confirm that Ollama remains available on this computer.
- Check that llama3.1:8b remains downloaded.
- Confirm that all five files remain inside data.
- Compare the reported Python line with the matching line in schemas.py, workflow.py, or app.py.
Use this prompt to diagnose the local preview safely. Help me debug the typed scope preview.
- Expand outputs in the existing Visual Studio Code sidebar.
- Expand demo-preview inside outputs.
- Select scope.json to inspect the generated preview.
- Confirm that THR-001 appears in exactly one scope item.
- Confirm that THR-002 appears in exactly one scope item.
- Confirm that THR-003 appears in exactly one scope item.
- Check that every scope item sets authorization_required to true.
You should see three typed scope items with synthetic boundaries and preserved threat IDs. That is the first working artifact from your local planning copilot.
Your copilot can now turn fixed synthetic threats into a validated scope artifact. Next, you will connect the planning stages and make the workflow stop at its human approval boundary.
Orchestrate Agents and Pause for Approval
Your typed scope preview now preserves every synthetic threat inside an authorization-required boundary. That is a solid checkpoint for the planning workflow.
The next risk is authorization drift. This step uses LangGraph to enforce the stage order before a human-in-the-loop interrupt stops evidence processing.
A local SQLite checkpoint preserves the paused thread. The same engagement ID can resume that exact state after approval.
In this step, get ready to:
- Build the source-grounded enrichment and safe guidance stages.
- Connect the workflow to a persistent human approval gate.
- Run the plan until it pauses before coverage processing.
Build the guarded workflow
The language model contributes source summaries and safe planning guidance. Deterministic Python code preserves identifiers throughout those stages.
The approval gate creates the authorization boundary. Coverage processing stays downstream from that interruption.
- Switch back to workflow.py from the scope preview.
- Keep the existing scope-preview behavior inside draft_scope.
- Update workflow.py until it matches the complete reference in the second tab below.
✔️ Awesome, I've got everything!
Your workflow now contains the enrichment, guidance, approval, coverage, reporting, and graph-building stages.
ⓧ I'd like to double check the full code
Compare your complete workflow.py file with this reference:
import json
from pathlib import Path
from typing import TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt
from ollama import chat
from pydantic import BaseModel
from schemas import (
ApprovalDecision,
AutomationResult,
CoverageItem,
CoverageReport,
EnrichedRisk,
EnrichmentBundle,
EnrichmentNarratives,
GuidanceBundle,
GuidanceNarratives,
ScopeDraft,
TestGuidance,
)
ROOT = Path(__file__).resolve().parent
DATA_DIR = ROOT / "data"
OUTPUT_DIR = ROOT / "outputs"
MODEL = "llama3.1:8b"
class WorkflowState(TypedDict, total=False):
engagement_id: str
scope: dict
enriched_risks: dict
guidance: dict
approved: bool
approval_note: str
coverage: dict
report_path: str
def load_json(path: Path):
return json.loads(path.read_text(encoding="utf-8"))
def write_model(path: Path, model: BaseModel) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(model.model_dump_json(indent=2), encoding="utf-8")
def engagement_dir(engagement_id: str) -> Path:
path = OUTPUT_DIR / engagement_id
path.mkdir(parents=True, exist_ok=True)
return path
def call_structured(schema: type[BaseModel], system_prompt: str, payload: dict) -> BaseModel:
schema_dict = schema.model_json_schema()
response = chat(
model=MODEL,
messages=[
{"role": "system", "content": system_prompt},
{
"role": "user",
"content": (
"Use only this input:\n"
+ json.dumps(payload, indent=2)
+ "\nReturn data matching this JSON schema:\n"
+ json.dumps(schema_dict, indent=2)
),
},
],
format=schema_dict,
options={"temperature": 0},
)
return schema.model_validate_json(response.message.content)
def draft_scope(state: WorkflowState) -> dict:
engagement_id = state["engagement_id"]
risks = load_json(DATA_DIR / "risk-table.json")
payload = {
"engagement_id": engagement_id,
"threat_model": (DATA_DIR / "threat-model.md").read_text(encoding="utf-8"),
"security_requirements": (DATA_DIR / "security-requirements.md").read_text(
encoding="utf-8"
),
"risk_table": risks,
}
scope = call_structured(
ScopeDraft,
(
"You are a defensive penetration-test scope planner. Create exactly one scope "
"item per threat_id. Copy all identifiers exactly. Keep every activity synthetic "
"and non-executing. State that authorization is required. Do not add targets, "
"payloads, exploit commands, or facts outside the supplied input."
),
payload,
)
scope = ScopeDraft.model_validate(
{**scope.model_dump(), "engagement_id": engagement_id}
)
expected_threats = {risk["threat_id"] for risk in risks}
actual_threats = {item.threat_id for item in scope.items}
if actual_threats != expected_threats or len(scope.items) != len(risks):
raise ValueError("Scope output did not preserve the supplied threat IDs exactly.")
write_model(engagement_dir(engagement_id) / "scope.json", scope)
return {"scope": scope.model_dump()}
def enrich_risks(state: WorkflowState) -> dict:
engagement_id = state["engagement_id"]
risks = load_json(DATA_DIR / "risk-table.json")
catalog = load_json(DATA_DIR / "reference-catalog.json")
payload = {
"scope": state["scope"],
"risks": risks,
"reference_catalog": catalog,
}
narratives = call_structured(
EnrichmentNarratives,
(
"You are a source-grounded security analyst. Return exactly one narrative for "
"each supplied risk_id. Summarize only the supplied catalog focus text and explain "
"why it matters to the supplied risk. Do not create identifiers, mappings, or "
"testing claims that are absent from the input."
),
payload,
)
narrative_by_risk = {item.risk_id: item for item in narratives.items}
expected_risks = {risk["risk_id"] for risk in risks}
if set(narrative_by_risk) != expected_risks or len(narratives.items) != len(risks):
raise ValueError("Enrichment output did not preserve the supplied risk IDs exactly.")
enriched = EnrichmentBundle(
items=[
EnrichedRisk(
risk_id=risk["risk_id"],
threat_id=risk["threat_id"],
source_refs=risk["source_refs"],
source_summary=narrative_by_risk[risk["risk_id"]].source_summary,
risk_rationale=narrative_by_risk[risk["risk_id"]].risk_rationale,
)
for risk in risks
]
)
write_model(engagement_dir(engagement_id) / "enrichment.json", enriched)
return {"enriched_risks": enriched.model_dump()}
def generate_guidance(state: WorkflowState) -> dict:
engagement_id = state["engagement_id"]
enriched = EnrichmentBundle.model_validate(state["enriched_risks"])
payload = {
"scope": state["scope"],
"enriched_risks": enriched.model_dump(),
"constraint": "Planning guidance only. No network activity or exploit payloads.",
}
narratives = call_structured(
GuidanceNarratives,
(
"You are a defensive test-planning engineer. Return exactly one guidance item for "
"each supplied risk_id. Write safe review and evidence-collection steps only. Do not "
"provide exploit payloads, commands, credential attacks, or instructions to contact "
"a system. Make manual limitations explicit."
),
payload,
)
narrative_by_risk = {item.risk_id: item for item in narratives.items}
expected_risks = {item.risk_id for item in enriched.items}
if set(narrative_by_risk) != expected_risks or len(narratives.items) != len(
enriched.items
):
raise ValueError("Guidance output did not preserve the supplied risk IDs exactly.")
guidance = GuidanceBundle(
items=[
TestGuidance(
risk_id=item.risk_id,
threat_id=item.threat_id,
source_refs=item.source_refs,
test_title=narrative_by_risk[item.risk_id].test_title,
safe_steps=narrative_by_risk[item.risk_id].safe_steps,
expected_evidence=narrative_by_risk[item.risk_id].expected_evidence,
manual_notes=narrative_by_risk[item.risk_id].manual_notes,
)
for item in enriched.items
]
)
write_model(engagement_dir(engagement_id) / "guidance.json", guidance)
return {"guidance": guidance.model_dump()}
def approval_gate(state: WorkflowState) -> dict:
scope = ScopeDraft.model_validate(state["scope"])
guidance = GuidanceBundle.model_validate(state["guidance"])
decision = interrupt(
{
"question": "Approve this synthetic scope before coverage processing?",
"engagement_id": state["engagement_id"],
"scope_items": len(scope.items),
"guidance_items": len(guidance.items),
},
response_schema=ApprovalDecision,
)
return {
"approved": decision.approved,
"approval_note": decision.note or "",
}
def compile_coverage(state: WorkflowState) -> dict:
engagement_id = state["engagement_id"]
risks = load_json(DATA_DIR / "risk-table.json")
if not state["approved"]:
coverage = CoverageReport(
engagement_id=engagement_id,
approved=False,
approval_note=state.get("approval_note", ""),
summary="Scope rejected. Simulated evidence was not processed.",
items=[],
)
write_model(engagement_dir(engagement_id) / "coverage.json", coverage)
return {"coverage": coverage.model_dump()}
results = [
AutomationResult.model_validate(item)
for item in load_json(DATA_DIR / "automation-results.json")
]
result_by_risk = {item.risk_id: item for item in results}
items: list[CoverageItem] = []
for risk in risks:
result = result_by_risk.get(risk["risk_id"])
if result is None:
status = "manual-required"
evidence_source = "none"
evidence = "No normalized automation result was supplied."
else:
status = result.status
evidence_source = result.source
evidence = result.evidence
items.append(
CoverageItem(
risk_id=risk["risk_id"],
threat_id=risk["threat_id"],
status=status,
evidence_source=evidence_source,
evidence=evidence,
source_refs=risk["source_refs"],
manual_required=status in {"partial", "manual-required", "blocked"},
)
)
ordered_statuses = [
"automated-covered",
"partial",
"manual-required",
"blocked",
"out-of-scope",
]
counts = {
status: sum(item.status == status for item in items)
for status in ordered_statuses
}
summary = ", ".join(
f"{status}={counts[status]}" for status in ordered_statuses if counts[status]
)
coverage = CoverageReport(
engagement_id=engagement_id,
approved=True,
approval_note=state.get("approval_note", ""),
summary=summary,
items=items,
)
write_model(engagement_dir(engagement_id) / "coverage.json", coverage)
return {"coverage": coverage.model_dump()}
def compile_report(state: WorkflowState) -> dict:
engagement_id = state["engagement_id"]
scope = ScopeDraft.model_validate(state["scope"])
enriched = EnrichmentBundle.model_validate(state["enriched_risks"])
guidance = GuidanceBundle.model_validate(state["guidance"])
coverage = CoverageReport.model_validate(state["coverage"])
catalog = {
item["id"]: item for item in load_json(DATA_DIR / "reference-catalog.json")
}
lines = [
f"# Pentest Planning and Coverage Report: {engagement_id}",
"",
"## Approval",
"",
f"Approved: {'Yes' if coverage.approved else 'No'}",
f"Note: {coverage.approval_note or 'No note supplied'}",
"",
"## Scope",
"",
]
for item in scope.items:
lines.extend(
[
f"### {item.scope_id}: {item.threat_id}",
"",
f"Asset: {item.asset}",
f"Objective: {item.test_objective}",
f"Boundaries: {', '.join(item.boundaries)}",
"",
]
)
lines.extend(["## Source-Grounded Risk Enrichment", ""])
for item in enriched.items:
lines.extend(
[
f"### {item.risk_id}: {item.threat_id}",
"",
item.source_summary,
"",
item.risk_rationale,
"",
"References:",
]
)
for ref in item.source_refs:
source = catalog[ref]
lines.append(f"- [{ref}: {source['title']}]({source['url']})")
lines.append("")
lines.extend(["## Test Guidance", ""])
for item in guidance.items:
lines.extend([f"### {item.risk_id}: {item.test_title}", ""])
for step in item.safe_steps:
lines.append(f"- {step}")
lines.extend(
[
"",
f"Expected evidence: {item.expected_evidence}",
f"Manual notes: {item.manual_notes}",
"",
]
)
lines.extend(["## Coverage", "", coverage.summary, ""])
for item in coverage.items:
lines.extend(
[
f"### {item.risk_id}: {item.status}",
"",
f"Threat: {item.threat_id}",
f"Evidence source: {item.evidence_source}",
f"Evidence: {item.evidence}",
f"Manual required: {'Yes' if item.manual_required else 'No'}",
"",
]
)
lines.extend(["## Manual Review Queue", ""])
manual_items = [item for item in coverage.items if item.manual_required]
if manual_items:
for item in manual_items:
lines.append(f"- {item.risk_id} / {item.threat_id}: {item.status}")
else:
lines.append("- No manual review items.")
lines.append("")
report_path = engagement_dir(engagement_id) / "report.md"
report_path.write_text("\n".join(lines), encoding="utf-8")
return {"report_path": str(report_path)}
def build_graph(checkpointer):
builder = StateGraph(WorkflowState)
builder.add_node("draft_scope", draft_scope)
builder.add_node("enrich_risks", enrich_risks)
builder.add_node("generate_guidance", generate_guidance)
builder.add_node("approval_gate", approval_gate)
builder.add_node("compile_coverage", compile_coverage)
builder.add_node("compile_report", compile_report)
builder.add_edge(START, "draft_scope")
builder.add_edge("draft_scope", "enrich_risks")
builder.add_edge("enrich_risks", "generate_guidance")
builder.add_edge("generate_guidance", "approval_gate")
builder.add_edge("approval_gate", "compile_coverage")
builder.add_edge("compile_coverage", "compile_report")
builder.add_edge("compile_report", END)
return builder.compile(checkpointer=checkpointer)
How the workflow enforces control
- The enrichment stage accepts model-written narratives only after every supplied risk ID is preserved.
- The guidance stage copies trusted threat IDs and source references from validated enrichment objects.
- The approval gate validates its input before interrupt pauses the graph.
- The graph places coverage compilation after the approval gate, so the paused run cannot process evidence.
- Save workflow.py.
Persist the approval thread
A checkpoint gives each graph thread a durable state. The thread_id binds that state to the engagement ID.
Your app.py file already sets LANGGRAPH_STRICT_MSGPACK before the checkpointer import. The plan command now opens the database before invoking the graph.
- Switch back to app.py from the preview command.
- Keep the existing LANGGRAPH_STRICT_MSGPACK setup at the top of the file.
- Update app.py until it matches the complete reference in the second tab below.
✔️ Awesome, I've got everything!
Your command-line app now opens the checkpoint database and routes the plan command into the graph.
ⓧ I'd like to double check the full code
Compare your complete app.py file with this reference:
import json
import os
import sqlite3
import sys
from pathlib import Path
os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command
from schemas import CoverageReport, EnrichmentBundle, GuidanceBundle, ScopeDraft
from workflow import DATA_DIR, ROOT, build_graph, draft_scope
def config_for(engagement_id: str) -> dict:
return {"configurable": {"thread_id": engagement_id}}
def open_graph():
connection = sqlite3.connect(
ROOT / "checkpoints.sqlite",
check_same_thread=False,
)
checkpointer = SqliteSaver(connection)
return connection, build_graph(checkpointer)
def preview(engagement_id: str) -> None:
result = draft_scope({"engagement_id": engagement_id})
scope = ScopeDraft.model_validate(result["scope"])
print(f"SCOPE PREVIEW: {len(scope.items)} items")
print(ROOT / "outputs" / engagement_id / "scope.json")
def plan(engagement_id: str) -> None:
connection, graph = open_graph()
try:
result = graph.invoke(
{"engagement_id": engagement_id},
config_for(engagement_id),
)
finally:
connection.close()
if "__interrupt__" not in result:
raise RuntimeError("Expected an approval interrupt, but the graph did not pause.")
print("STATUS: WAITING FOR HUMAN APPROVAL")
print(f"Thread: {engagement_id}")
print("Review scope.json, enrichment.json, and guidance.json before resuming.")
def resume(engagement_id: str, approved: bool) -> None:
connection, graph = open_graph()
note = (
"Approved for synthetic evidence only"
if approved
else "Rejected during local review"
)
try:
result = graph.invoke(
Command(resume={"approved": approved, "note": note}),
config_for(engagement_id),
)
finally:
connection.close()
print(f"APPROVED: {approved}")
print(result["report_path"])
def audit(engagement_id: str) -> None:
output_dir = ROOT / "outputs" / engagement_id
risks = json.loads((DATA_DIR / "risk-table.json").read_text(encoding="utf-8"))
catalog = json.loads(
(DATA_DIR / "reference-catalog.json").read_text(encoding="utf-8")
)
scope = ScopeDraft.model_validate_json(
(output_dir / "scope.json").read_text(encoding="utf-8")
)
enrichment = EnrichmentBundle.model_validate_json(
(output_dir / "enrichment.json").read_text(encoding="utf-8")
)
guidance = GuidanceBundle.model_validate_json(
(output_dir / "guidance.json").read_text(encoding="utf-8")
)
coverage = CoverageReport.model_validate_json(
(output_dir / "coverage.json").read_text(encoding="utf-8")
)
expected_threats = {item["threat_id"] for item in risks}
expected_risks = {item["risk_id"] for item in risks}
known_refs = {item["id"] for item in catalog}
checks = {
"scope threats": {item.threat_id for item in scope.items} == expected_threats,
"enrichment risks": {item.risk_id for item in enrichment.items}
== expected_risks,
"guidance risks": {item.risk_id for item in guidance.items} == expected_risks,
"coverage risks": {item.risk_id for item in coverage.items} == expected_risks,
"reference IDs": all(
set(item.source_refs) <= known_refs for item in enrichment.items
),
"coverage evidence": all(bool(item.evidence.strip()) for item in coverage.items),
"approval": coverage.approved,
}
failed = [name for name, passed in checks.items() if not passed]
if failed:
raise RuntimeError("Traceability audit failed: " + ", ".join(failed))
print("TRACEABILITY AUDIT: PASS")
print(f"Threats: {len(expected_threats)}")
print(f"Risks: {len(expected_risks)}")
print(f"References: {len(known_refs)}")
def usage() -> None:
script = Path(sys.argv[0]).name
print(f"Usage: {script} preview|plan|approve|reject|audit <engagement-id>")
def main() -> None:
if len(sys.argv) != 3:
usage()
raise SystemExit(2)
command, engagement_id = sys.argv[1], sys.argv[2]
if command == "preview":
preview(engagement_id)
elif command == "plan":
plan(engagement_id)
elif command == "approve":
resume(engagement_id, True)
elif command == "reject":
resume(engagement_id, False)
elif command == "audit":
audit(engagement_id)
else:
usage()
raise SystemExit(2)
if __name__ == "__main__":
main()
How the checkpoint is opened
- The environment setting restricts checkpoint deserialization before SqliteSaver is imported.
- The open_graph helper connects to checkpoints.sqlite with thread checking disabled for graph access.
- The config_for helper uses the engagement ID as the persistent thread_id.
- The plan command rejects any run that reaches the end without the expected interruption.
- Save app.py.
Run the plan into the approval gate
This run uses demo-approved as both the engagement ID and the checkpoint thread. The model may take a few minutes to generate all three planning artifacts.
Before you run the plan, do you expect the graph to reach coverage processing or pause first?
- Start the persisted planning thread by running this command in the PowerShell terminal from earlier:
.\.venv\Scripts\python.exe app.py plan demo-approved
What does this run do?
- The graph drafts a fresh scope for demo-approved.
- The enrichment stage grounds each risk in the bundled reference catalog.
- The guidance stage writes safe planning instructions without contacting a system.
- The approval node records an interrupt inside checkpoints.sqlite before returning control to the command-line app.
You should see STATUS: WAITING FOR HUMAN APPROVAL in the terminal. This confirms that the authorization boundary stopped the graph on its main path.
Did the planning run stop early?
Confirm that Ollama is still running locally. A stopped local model prevents the structured planning stages from completing.
Check your workflow.py indentation against the full reference above. Python must keep each validation block inside its matching function.
Ask for help with the exact terminal output if the run does not reach the approval message: Help me diagnose why my LangGraph plan command did not pause at approval.
- In the VS Code Explorer sidebar, expand the outputs folder.
- Expand the demo-approved folder.
- Confirm that scope.json, enrichment.json, and guidance.json are listed.
- Return to the project folder in the Explorer sidebar.
- Confirm that checkpoints.sqlite is listed.
- Confirm that coverage.json is absent from outputs/demo-approved.
- Confirm that report.md is absent from outputs/demo-approved.
You have reached the intended stop. The planning artifacts exist, while evidence classification remains blocked behind human approval.
Your graph is paused with its state preserved in SQLite. Next, you will resume this exact thread with an explicit decision.
Resume the Approved Workflow
Your LangGraph workflow has reached the authorization boundary with its scope plus guidance safely persisted. The paused graph has no authority to process evidence until you supply a human decision.
You now act as the approver for the synthetic engagement. The existing SQLite checkpoint lets the exact demo-approved thread continue into deterministic coverage processing.
In this step, get ready to:
- Resume the persisted thread with explicit human approval.
- Verify the deterministic JSON coverage statuses.
- Inspect the completed Markdown report for unresolved work.
Review and resume the approval gate
In app.py, resume() sends Command(resume={"approved": True, "note": "Approved for synthetic evidence only"}) to the existing thread. The same demo-approved thread ID tells LangGraph to continue from the stored interrupt.
This approval is safe to grant because it releases only the bundled synthetic evidence inside this local project.
- Select outputs/demo-approved/scope.json in the Explorer sidebar in Visual Studio Code.
- Confirm every scope item has authorization_required set to true.
- Switch back to the PowerShell terminal from the previous step.
Before you resume the thread, do you expect the workflow to restart from scope generation or continue after the approval gate?
- Resume the persisted demo-approved thread by running this command:
.\.venv\Scripts\python.exe app.py approve demo-approved
What Does This Approval Do?
- The approve command routes the decision through resume().
- The demo-approved value selects the persisted thread in checkpoints.sqlite.
- The Command object supplies the explicit approval decision to the interrupted node.
- The graph continues through compile_coverage before passing its result to compile_report.
The key safeguard just proved itself: the workflow continued only after your explicit approval.
You'll see APPROVED: True. The next line ends with outputs\demo-approved\report.md.
Thread Does Not Resume?
- Confirm checkpoints.sqlite is still in the same project folder as app.py.
- Confirm the engagement ID remains demo-approved.
- Return to Step 3 if the saved approval interrupt is missing.
Help me diagnose why the paused demo-approved thread will not resume.
Inspect deterministic coverage
The compile_coverage node loads the simulated records from data/automation-results.json. Pydantic validates each AutomationResult before the node creates a CoverageReport.
The model has no control over the approval decision. It also has no control over the evidence or coverage statuses.
- Switch back to Visual Studio Code from earlier.
- Expand the outputs folder in the Explorer sidebar.
- Expand the demo-approved folder.
- Select coverage.json.
What Should the Coverage Show?
- The RISK-001 status is partial.
- The RISK-001 value for manual_required is true.
- The RISK-002 status is automated-covered.
- The RISK-002 value for manual_required is false.
- The RISK-003 status is blocked.
- The RISK-003 value for manual_required is true.
- The summary value is automated-covered=1, partial=1, blocked=1.
Verify the coverage report
The compile_report node combines the approved artifacts with the validated coverage. Its Manual Review Queue section keeps unresolved risks visible for a human handoff.
Before you open the report, which risks do you expect to find in the manual review queue?
- Select report.md in the expanded demo-approved folder.
- Locate the ## Approval section.
- Locate the ## Coverage section.
- Locate the ## Manual Review Queue section.
You'll see Approved: Yes in the approval section. The note reads Approved for synthetic evidence only.
The coverage summary reads automated-covered=1, partial=1, blocked=1. The manual review queue lists RISK-001 plus RISK-003.
Why Do Unresolved Items Remain?
A partial status preserves the limits of incomplete evidence. A blocked status preserves the absence of approved execution evidence.
Keeping both statuses visible prevents the final report from overstating coverage.
Your persisted thread now carries an explicit approval into evidence-based reporting. Next, you'll audit every threat plus risk link across the completed engagement.
Audit End-to-End Traceability
Your approved workflow now has a persisted decision. It also has a coverage JSON artifact and a traceable Markdown report built from simulated evidence.
A polished report cannot be trusted if identifiers disappear between stages. This step adds a final audit that checks the complete evidence chain before you treat the report as finished.
In this step, get ready to:
- Compare threat IDs with the synthetic risk table.
- Verify risk IDs across enrichment, guidance, and coverage.
- Run the audit to prove every reference and evidence record remains traceable.
Compare IDs across artifacts
The audit uses Pydantic to validate each generated artifact before comparing its identifiers. Invalid artifact structures therefore fail before they can produce a misleading traceability result.
- In app.py, find the line that starts def usage().
- Add the audit loader directly above that line by pasting this code:
def audit(engagement_id: str) -> None:
output_dir = ROOT / "outputs" / engagement_id
risks = json.loads((DATA_DIR / "risk-table.json").read_text(encoding="utf-8"))
catalog = json.loads(
(DATA_DIR / "reference-catalog.json").read_text(encoding="utf-8")
)
scope = ScopeDraft.model_validate_json(
(output_dir / "scope.json").read_text(encoding="utf-8")
)
enrichment = EnrichmentBundle.model_validate_json(
(output_dir / "enrichment.json").read_text(encoding="utf-8")
)
guidance = GuidanceBundle.model_validate_json(
(output_dir / "guidance.json").read_text(encoding="utf-8")
)
coverage = CoverageReport.model_validate_json(
(output_dir / "coverage.json").read_text(encoding="utf-8")
)
expected_threats = {item["threat_id"] for item in risks}
expected_risks = {item["risk_id"] for item in risks}
known_refs = {item["id"] for item in catalog}
What Does This Code Load?
- The risk table supplies the expected threat IDs and risk IDs.
- The reference catalog supplies the complete set of allowed source IDs.
- The four schema calls validate the scope, enrichment, guidance, and coverage artifacts.
- The three sets create stable collections for exact comparisons.
- Save app.py.
- Confirm the new audit() function ends with the known_refs set in your editor.
Audit Loader in the Wrong Place?
Keep audit() at the leftmost indentation level. Make sure it appears above usage() instead of inside another function.
Use this if the function boundaries look unclear: Help me check where the audit function belongs in app.py.
Verify references and evidence
Identifier equality proves that each stage covers the expected records. The audit also confirms that every source resolves to the bundled OWASP Web Security Testing Guide or MITRE ATT&CK® catalog.
- In audit(), find the line that defines known_refs.
- Add the audit checks directly below that line by pasting this code:
checks = {
"scope threats": {item.threat_id for item in scope.items} == expected_threats,
"enrichment risks": {item.risk_id for item in enrichment.items}
== expected_risks,
"guidance risks": {item.risk_id for item in guidance.items} == expected_risks,
"coverage risks": {item.risk_id for item in coverage.items} == expected_risks,
"reference IDs": all(
set(item.source_refs) <= known_refs for item in enrichment.items
),
"coverage evidence": all(bool(item.evidence.strip()) for item in coverage.items),
"approval": coverage.approved,
}
failed = [name for name, passed in checks.items() if not passed]
if failed:
raise RuntimeError("Traceability audit failed: " + ", ".join(failed))
print("TRACEABILITY AUDIT: PASS")
print(f"Threats: {len(expected_threats)}")
print(f"Risks: {len(expected_risks)}")
print(f"References: {len(known_refs)}")
What Do These Checks Prove?
- The first four checks compare expected IDs with the IDs preserved in each generated artifact.
- The reference check rejects any enrichment source that is absent from the bundled catalog.
- The evidence check rejects empty coverage evidence.
- The approval check keeps the human decision authoritative at the final checkpoint.
- Save app.py.
- Confirm the checks dictionary contains seven named audit conditions.
Seeing Indentation Problems?
Keep the entire checks section indented inside audit(). Align checks with expected_threats.
Use this if a comparison is outside the function: Help me fix the indentation of the checks inside audit().
Run the traceability audit
The command dispatcher needs one final branch that connects the audit command to audit(). The completed command then evaluates the approved engagement without contacting any target.
- In main(), find this command branch:
elif command == "reject":
resume(engagement_id, False)
Why Use This Branch as the Marker?
The reject branch is the last existing engagement command. The audit branch belongs directly below it so every supported command remains in one dispatcher.
- Replace that branch with this expanded command routing:
elif command == "reject":
resume(engagement_id, False)
elif command == "audit":
audit(engagement_id)
What Does This Route Do?
The new branch passes the supplied engagement ID into audit(). This lets the same audit work with the existing demo-approved output folder.
- Save app.py.
Before you run the audit, do you think all seven checks will agree on the completed approved engagement?
- Audit the completed demo-approved engagement by running this command in the PowerShell terminal from earlier:
.\.venv\Scripts\python.exe app.py audit demo-approved
What Should You See?
You should see TRACEABILITY AUDIT: PASS followed by Threats: 3, Risks: 3, and References: 4.
These counts prove that all three synthetic threats and risks remain connected to the four bundled source references.
Audit Did Not Pass?
- Check that outputs/demo-approved still contains scope.json, enrichment.json, guidance.json, and coverage.json.
- Check that you ran the command with the existing demo-approved engagement ID.
- Read the named failed check to identify which artifact lost an ID, reference, evidence value, or approval state.
Use the failed check for focused help: Help me diagnose my failed traceability audit.
You have closed the trust loop. The approved synthetic engagement now proves its identifiers, references, evidence, and human decision survived every stage.
✔️ Awesome, I've got everything!
Great work. Save app.py and keep the passing audit output as proof of the complete evidence chain.
ⓧ I'd like to double check the full code
Compare your app.py with the complete command-line implementation below.
import json
import os
import sqlite3
import sys
from pathlib import Path
os.environ.setdefault("LANGGRAPH_STRICT_MSGPACK", "true")
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import Command
from schemas import CoverageReport, EnrichmentBundle, GuidanceBundle, ScopeDraft
from workflow import DATA_DIR, ROOT, build_graph, draft_scope
def config_for(engagement_id: str) -> dict:
return {"configurable": {"thread_id": engagement_id}}
def open_graph():
connection = sqlite3.connect(
ROOT / "checkpoints.sqlite",
check_same_thread=False,
)
checkpointer = SqliteSaver(connection)
return connection, build_graph(checkpointer)
def preview(engagement_id: str) -> None:
result = draft_scope({"engagement_id": engagement_id})
scope = ScopeDraft.model_validate(result["scope"])
print(f"SCOPE PREVIEW: {len(scope.items)} items")
print(ROOT / "outputs" / engagement_id / "scope.json")
def plan(engagement_id: str) -> None:
connection, graph = open_graph()
try:
result = graph.invoke(
{"engagement_id": engagement_id},
config_for(engagement_id),
)
finally:
connection.close()
if "__interrupt__" not in result:
raise RuntimeError("Expected an approval interrupt, but the graph did not pause.")
print("STATUS: WAITING FOR HUMAN APPROVAL")
print(f"Thread: {engagement_id}")
print("Review scope.json, enrichment.json, and guidance.json before resuming.")
def resume(engagement_id: str, approved: bool) -> None:
connection, graph = open_graph()
note = (
"Approved for synthetic evidence only"
if approved
else "Rejected during local review"
)
try:
result = graph.invoke(
Command(resume={"approved": approved, "note": note}),
config_for(engagement_id),
)
finally:
connection.close()
print(f"APPROVED: {approved}")
print(result["report_path"])
def audit(engagement_id: str) -> None:
output_dir = ROOT / "outputs" / engagement_id
risks = json.loads((DATA_DIR / "risk-table.json").read_text(encoding="utf-8"))
catalog = json.loads(
(DATA_DIR / "reference-catalog.json").read_text(encoding="utf-8")
)
scope = ScopeDraft.model_validate_json(
(output_dir / "scope.json").read_text(encoding="utf-8")
)
enrichment = EnrichmentBundle.model_validate_json(
(output_dir / "enrichment.json").read_text(encoding="utf-8")
)
guidance = GuidanceBundle.model_validate_json(
(output_dir / "guidance.json").read_text(encoding="utf-8")
)
coverage = CoverageReport.model_validate_json(
(output_dir / "coverage.json").read_text(encoding="utf-8")
)
expected_threats = {item["threat_id"] for item in risks}
expected_risks = {item["risk_id"] for item in risks}
known_refs = {item["id"] for item in catalog}
checks = {
"scope threats": {item.threat_id for item in scope.items} == expected_threats,
"enrichment risks": {item.risk_id for item in enrichment.items}
== expected_risks,
"guidance risks": {item.risk_id for item in guidance.items} == expected_risks,
"coverage risks": {item.risk_id for item in coverage.items} == expected_risks,
"reference IDs": all(
set(item.source_refs) <= known_refs for item in enrichment.items
),
"coverage evidence": all(bool(item.evidence.strip()) for item in coverage.items),
"approval": coverage.approved,
}
failed = [name for name, passed in checks.items() if not passed]
if failed:
raise RuntimeError("Traceability audit failed: " + ", ".join(failed))
print("TRACEABILITY AUDIT: PASS")
print(f"Threats: {len(expected_threats)}")
print(f"Risks: {len(expected_risks)}")
print(f"References: {len(known_refs)}")
def usage() -> None:
script = Path(sys.argv[0]).name
print(f"Usage: {script} preview|plan|approve|reject|audit <engagement-id>")
def main() -> None:
if len(sys.argv) != 3:
usage()
raise SystemExit(2)
command, engagement_id = sys.argv[1], sys.argv[2]
if command == "preview":
preview(engagement_id)
elif command == "plan":
plan(engagement_id)
elif command == "approve":
resume(engagement_id, True)
elif command == "reject":
resume(engagement_id, False)
elif command == "audit":
audit(engagement_id)
else:
usage()
raise SystemExit(2)
if __name__ == "__main__":
main()
What Should Match?
Your full file should contain the audit loader, seven checks, failure handling, passing counts, and the audit command route. Keep the existing preview, planning, approval, rejection, and checkpoint logic unchanged.
Secret mission
Export the Manual Test Queue
Your coverage report identifies unresolved risks inside a larger engagement artifact. In this mission, you will export those risks as a focused manual test queue while leaving the trusted workflow unchanged.
Clean Up Your Resources
Clean Up Your Resources
Your copilot runs entirely on your Windows computer, so leaving its resources idle creates no usage charges. Choose whether to keep everything available, pause your work, or remove the local resources.
Resources you used:
- The Python environment at .venv inside the local project folder.
- The synthetic inputs under data inside the local project folder.
- The saved workflow state in checkpoints.sqlite inside the local project folder.
- The generated engagement artifacts under outputs/demo-approved inside the local project folder.
- The application source files inside the local project folder.
- The local llama3.1:8b model managed by Ollama.
Keep everything running
No action is needed. Choose this option if you want to demonstrate the approved workflow again or continue building local security-planning tools.
- Leave the local project folder in its current location.
- Leave the llama3.1:8b model available for future local runs.
- Keep the completed demo-approved thread as a resumable workflow example.
Pause - I'll come back to this later
Pausing closes your project tools while keeping every file on disk. No metered service continues running.
- Close the PowerShell window used for the project.
- Close Visual Studio Code.
- Leave the local project folder in place for your next session.
- Leave the llama3.1:8b model on disk to avoid downloading it again.
Delete - I don't want to use this again
Deletion removes the downloaded model plus every project artifact. Your Ollama application remains installed for other local models.
- Remove the downloaded model from Ollama by running this command in PowerShell:
ollama rm llama3.1:8b
What Does This Command Remove?
This removes the 4.9 GB model files for llama3.1:8b from Ollama's local model storage. It leaves the Ollama application installed.
- Confirm that PowerShell returns to its prompt without reporting an error.
Model Still Present?
- Close any terminal session that is still running the model.
- Retry the model removal command.
Get help with the removal command.
- Close Visual Studio Code.
- Close any PowerShell window that is using the local project folder.
- Locate the local project folder in File Explorer by finding the folder that contains app.py.
- Use File Explorer to delete the local project folder.
- Confirm that the local project folder no longer appears in its original location.
Nice Work!
Nice Work!
You did it! You built a local pentest planning copilot that turns synthetic security inputs into an approved coverage report without contacting a real target.
You've learned how to:
- Generate a typed pentest scope with Pydantic schemas. Your local Ollama model produced structured planning artifacts while deterministic checks preserved the supplied threat IDs.
- Orchestrate a human-approved workflow with LangGraph. The workflow paused at an authorization gate before resuming the same thread from a SQLite checkpoint.
- Prove end-to-end traceability across scope, enrichment, guidance, source references, simulated evidence, and coverage. The final audit confirmed that all three threats and risks remained linked.
- Secret Mission: Exported a manual test queue that preserves evidence for partial and blocked risks. The queue excludes work already covered by simulated automation.
Ready to quiz yourself?