to-note-at-each-step
Step 1 — The app is created
app = FastAPI(title="data-patch-agent", version="2.0")
This one line creates the entire web server. app is the object everything else hangs off. When uvicorn starts, it runs this file and starts listening for HTTP requests on port 8000.
Step 2 — A request arrives at /webhook
@app.post("/webhook")
async def webhook(request: Request):
@app.post("/webhook") tells FastAPI: when someone sends a POST request to /webhook, run this function. Shortcut sends one every time any event happens in your workspace — a comment posted, a story updated, anything.
Step 3 — Read the raw bytes and verify HMAC
body = await request.body()
sig = request.headers.get("X-Shortcut-Hmac-SHA256", "")
if not _verify_hmac(body, sig):
raise HTTPException(status_code=401, detail="Invalid HMAC signature")
We read the request body as raw bytes first. Not JSON yet — raw (i.e. we do not request.json() the body yet). This is critical because HMAC must operate on the exact bytes Shortcut sent. If we parsed to JSON first, Python might reorder keys or strip whitespace and the signature would no longer match.
★ HMAC
HMAC = Hash-based Message Authentication Code. It's a way to prove that a message came from someone who knows a shared secret, without transmitting the secret itself.
_verify_hmac() recomputes the expected signature using your SHORTCUT_WEBHOOK_SECRET and compares it to what Shortcut sent. If they don't match → 401, stop everything.
def _verify_hmac(payload: bytes, signature: str) -> bool:
expected = hmac.new(
config.SHORTCUT_WEBHOOK_SECRET.encode(),
payload,
hashlib.sha256,
).hexdigest()
return hmac.compare_digest(expected, signature.strip())
hmac.compare_digest instead of == — prevents timing attacks. == stops comparing the moment it finds a mismatched character. A patient attacker could measure response times to guess the secret one character at a time. compare_digest always takes the same amount of time regardless of how many characters matched.
Step 4 — Parse the event and filter to comments only
event = await request.json()
for action in event.get("actions", []):
if action.get("entity_type") != "story-comment":
continue
if action.get("action") not in ("create", "update"):
continue
Now we parse the raw bytes as JSON. A Shortcut event can contain multiple actions — a single user interaction might trigger "story updated + comment created" in one payload. We loop through all of them and skip everything that isn't a comment being created or edited.
Step 5 — Extract the fields we need
author_id = action.get("author_id", "")
text = (action.get("text") or "").strip()
story_id = action.get("story_id") or event.get("primary_id")
sc_id = f"SC-{story_id}"
if author_id == config.BOT_MEMBER_ID:
continue
trigger_comment_id = action.get("id") # ID of THIS comment
parent_id = action.get("parent_id") # None if top-level
_handle_comment(sc_id, text, parent_id, trigger_comment_id)
author_id— who posted the commenttext— the comment textstory_id— the numeric Shortcut story ID → we prefix it to get"SC-149949"trigger_comment_id— the ID of this specific comment. If the engineer posts@agentbot clarify, this becomes the permanent thread root — all bot replies will go into this thread for the lifetime of the sessionparent_id— if this is a reply to another comment, this is that comment's ID.Nonemeans it's a top-level comment
The bot-filter: if author_id == BOT_MEMBER_ID, skip. Without this the bot would react to its own comments and loop forever.
Then we hand off to _handle_comment().
Step 6 — _handle_comment() routes by session state
def _handle_comment(sc_id, text, parent_id, trigger_comment_id) -> None:
session = db.get_session(sc_id)
First thing: look up whether a session already exists for this story in the database.
if session is None:
if "@agentbot clarify" in text.lower():
_run_clarify(sc_id, trigger_comment_id)
return
No session → only act if the comment contains "@agentbot clarify". Everything else is ignored. trigger_comment_id is passed in so _run_clarify can store it as the thread root.
status = session["status"]
if status == "awaiting_extraction":
return
Session exists, waiting for SQL extraction → ignore all comments. The engineer needs to run 01_extract.sql and submit results via GitLab CI. No comment can advance this state.
if status == "awaiting_pm":
thread_comment_id = session.get("thread_comment_id")
if thread_comment_id is not None and parent_id != thread_comment_id:
return
if not db.update_status(sc_id, "drafting", "awaiting_pm"):
return
try:
_preprocess_and_advance(sc_id)
except Exception:
db.update_status(sc_id, "awaiting_pm", "drafting")
logger.exception("Error processing webhook for %s", sc_id)
raise
return
Session exists, waiting for PM answers → only respond if the comment is a reply inside the engineer's original thread (parent_id == thread_comment_id). Top-level comments and replies to other comments are silently ignored.
Then set status to "drafting" atomically — this acts as a processing lock so if two webhooks arrive at the same time, only one proceeds. update_status does UPDATE WHERE status='awaiting_pm' — the second webhook finds the status already changed, gets rowcount=0, returns False, and skips. If anything crashes, the except block rolls status back to awaiting_pm.
if status == "done":
shortcut_client.post_comment(
sc_id,
"This session is already complete — the patch scripts have been posted above.",
)
Done → politely decline.
Step 7 — Return 200 always
return {"ok": True}
The webhook function always returns 200 to Shortcut. If we returned anything else, Shortcut would retry the webhook — potentially processing the same comment twice. We return 200 and handle errors internally.
Step 8 — _run_clarify() is entered
def _run_clarify(sc_id: str, thread_comment_id: int) -> None:
We arrive here only once — when the engineer posts @agentbot clarify on a story with no existing session. thread_comment_id is the numeric ID of that exact comment. It becomes the permanent thread root and never changes for the life of the session.
Step 9 — Create the session and lock in the thread root
db.create_session(sc_id)
update_thread_comment_id(sc_id, thread_comment_id)
create_session writes a new row to patch_sessions:
INSERT INTO patch_sessions
(sc_id, status, clarify_history, hits, extraction_rounds, draft)
VALUES ('SC-149949', 'awaiting_extraction', '[]', '[]', '[]', 'null')
ON CONFLICT (sc_id) DO NOTHING
Status starts as awaiting_extraction and stays there until the end of _run_clarify. The ON CONFLICT DO NOTHING makes it safe if somehow called twice — no duplicate row, no crash.
update_thread_comment_id immediately writes the engineer's comment ID into patch_sessions.thread_comment_id. This is the only time this column is ever written. From here on every bot reply goes into this thread (parent_id=thread_comment_id), and every incoming PM comment is filtered by it.
Step 10 — Fetch the Shortcut story
story = shortcut_client.fetch_story(sc_id)
Calls GET /api/v3/stories/149949 with your SHORTCUT_API_TOKEN. Returns a cleaned dict with the fields the pipeline uses: title, description, comments (normalised to {id, author_id, text, created_at, updated_at}), labels, story_type, and more. Shortcut internals we don't need are stripped. The full raw response is kept under _raw for debugging.
Step 11 — Embed the story text
story_text = " ".join([
story.get("title", ""),
story.get("description", "") or "",
])
vectors, _ = call_embed([story_text])
Title + description are concatenated into one string and sent to text-embedding-3-small. call_embed returns (vectors, usage) — vectors is a list of one 1536-float list. Usage is discarded with _. Cost is logged to stderr and llm_costs.jsonl automatically inside call_embed.
Step 12 — RAG search: find the most similar past patches
hits = pgvector_client.search(vectors[0], top_k=5)
Takes our 1536-float vector and runs a cosine similarity search against the patch_corpus table:
SELECT id, 1 - (embedding <=> query_vector) AS score, payload
FROM patch_corpus
ORDER BY embedding <=> query_vector
LIMIT 5
<=> is pgvector's cosine distance operator. 1 - distance flips it to similarity (1.0 = identical, 0.0 = unrelated). Returns the top 5 as:
[{"id": int, "score": float, "payload": {...}}, ...]
Each payload contains story_id, title, description, labels, and script_content (all SQL from that past story concatenated). These become the few-shot examples injected into the clarify prompt.
Step 13 — Build the clarify prompt
messages = build_clarify_messages(story, hits)
Reads four files from disk:
- clarify.md — the main LLM instructions
- rules.md — business rules
- templates.md — template definitions A–F
- glossary.md — domain terms
Scans story text for known database table names and injects a schema block if any are found.
System message = clarify prompt + rules + templates + glossary + schema block.
User message = story title, labels, description, comments (capped at 4000 chars total) + the 5 past patches formatted as markdown, SQL snippets capped at 2000 chars each:
### Past patch 1 — SC-138609 (similarity 0.847)
**Title:** Patch EACases ApprovalRequest
**Labels:** template-B
**SQL content:**
```sql
UPDATE ...
Returns `[{"role": "system", "content": ...}, {"role": "user", "content": ...}]`.
---
## Step 14 — Call Sonnet
```python
response, _ = call_chat(messages, model=config.CHAT_MODEL)
Sends the two messages to azure.claude-sonnet-4-5, temperature=0.0, max_tokens=4096. Returns (text, usage). text is the raw string the model produced.
Known issue: max_tokens=4096 can truncate a long response mid-JSON, causing a parse failure with no retry logic in V2.
Step 15 — Parse the response
result = parse_clarify_result(response)
def parse_clarify_result(text: str) -> ClarifyResult:
cleaned = strip_json_fences(text) # strips ```json ... ``` if present
return ClarifyResult.model_validate_json(cleaned)
ClarifyResult is a Pydantic model. If the JSON is valid we get a typed object with:
inferred_template— which of A–F this story isconfidence— how confident the model isconfirmed_values— dict of field values already clear from the story textopen_questions— list of{id, question, impact}for things still unclearextract_sql— the01_extract.sqlthe engineer needs to run
If the JSON is malformed → ValidationError → propagates up to _handle_comment's except block → status rolls back to awaiting_pm → logged.
★ Pydantic
Pydantic is a Python library for data validation. A Pydantic model is a class that defines what shape a piece of data must have — field names, types, and allowed values.
class ClarifyResult(BaseModel):
inferred_template: ClarifyTemplate # must be "A", "B", "C", "D", "E", "F", or "novel"
confidence: Confidence # must be "high", "medium", or "low"
confirmed_values: dict[str, str] # must be a dict of strings
open_questions: list[OpenQuestion] # must be a list of OpenQuestion objects
extract_sql: str # must be a string
...
Without Pydantic you'd have to manually validate every field:
data = json.loads(text)
if "extract_sql" not in data:
raise ValueError("missing extract_sql")
if not isinstance(data["extract_sql"], str):
raise ValueError("extract_sql must be a string")
# ... for every field
Step 16 — Persist and post
db.append_clarify_result(sc_id, result, hits)
Appends to patch_sessions.clarify_history (a JSONB array — multiple clarify rounds accumulate here):
[{
"round": 1,
"generated_at": "2026-07-23T...",
"result": { "inferred_template": "B", "extract_sql": "SELECT ...", ... }
}]
Also overwrites patch_sessions.hits with the current top-5 results (used later if reclarify is needed).
shortcut_client.post_comment(
sc_id,
f"Here is the extraction SQL...\n\n```sql\n{result.extract_sql}\n```\n\n"
f"_Reply to this thread with your answers._",
parent_id=thread_comment_id,
)
Posts the extraction SQL to Shortcut as a reply to the engineer's comment. This is the first message in the thread. All future bot messages go into this same thread.
Step 17 — _preprocess_and_advance() is entered
This is called every time a PM replies in the thread (status was awaiting_pm). Its job: look at everything accumulated so far and decide what to do next — ask more questions, reclarify, or go straight to draft.
Step 18 — Load session state and story
session = db.get_session(sc_id)
story = shortcut_client.fetch_story(sc_id)
thread_comment_id = session.get("thread_comment_id")
Re-fetches both. The story fetch is important here — we need the latest comments, since the PM has just replied.
user_comments = [
c for c in story.get("comments", [])
if c.get("author_id") != config.BOT_MEMBER_ID
]
Filters down to only human comments. The bot's own posted messages (the extraction SQL, the questions) are stripped out — the Haiku model only needs to read what humans said.
clarify_result = session["clarify_history"][-1]["result"]
extraction_rounds = session["extraction_rounds"]
[-1] gets the most recent clarify round in case there were multiple. extraction_rounds is the list of raw data submitted by GitLab CI — may be empty if the PM is replying before extraction was run.
Step 19 — The Haiku preprocess call
messages = build_preprocess_messages(clarify_result, user_comments, extraction_rounds)
response, _ = call_chat(messages, model=config.CHEAP_CHAT_MODEL)
ctx = parse_session_context(response)
build_preprocess_messages assembles a user message containing four sections:
## Template
B
## Confirmed values
form_type: IHQ-DEI
## Open questions
Q1: Which iteration does this apply to?
Q2: Is this an amendment or new record?
## PM/engineer comments from Shortcut (chronological)
[2026-07-23T10:00:00Z] iteration 47, no amendments
## Extraction data
Round 1 (submitted 2026-07-23T09:55:00Z):
id | form_master_id | ...
The system message is preprocess.md — instructions telling Haiku how to map comments and extraction data onto the open questions.
parse_session_context parses the response into a SessionContext object:
template— confirmed template letterconfirmed_values— field values known with certaintyresolved—{question_id: synthesised answer}— what the PM answeredstill_open— question IDs with no answer yetextraction_summary— plain-English summary of what the extraction data showedneeds_reclarify— True if something contradicts the original template assumptionreclarify_reason— explanation if so
This is a cheap Haiku call — we're not asking it to write SQL, just to read text and classify answers. Sonnet is saved for the heavier clarify and draft calls.
Step 20 — Outcome A: Reclarify needed
if ctx.needs_reclarify:
messages = build_clarify_messages(story, session["hits"])
response, _ = call_chat(messages, model=config.CHAT_MODEL)
new_result = parse_clarify_result(response)
db.append_clarify_result(sc_id, new_result, session["hits"])
db.update_status(sc_id, "awaiting_extraction", "drafting")
shortcut_client.post_comment(sc_id, f"...{new_result.extract_sql}...", parent_id=thread_comment_id)
return
The PM's answers revealed the original template assumption was wrong — e.g. we assumed template B but it's actually template D. We run the full Sonnet clarify call again with the updated story, append a new entry to clarify_history (round 2), post a fresh extract_sql into the thread, and set status back to awaiting_extraction. The whole cycle can repeat.
Step 21 — Outcome B: Still open questions
if ctx.still_open:
open_q_map = {q["id"]: q["question"] for q in clarify_result.get("open_questions", [])}
q_lines = "\n".join(f"**{q_id}**: {open_q_map.get(q_id, q_id)}" for q_id in ctx.still_open)
shortcut_client.post_comment(sc_id, f"Still waiting on...\n\n{q_lines}", parent_id=thread_comment_id)
db.update_status(sc_id, "awaiting_pm", "drafting")
return
Some questions answered, but not all. open_q_map looks up the full question text from the original clarify result using the ID (e.g. "Q1" → "Which iteration does this apply to?"). We post only the remaining unanswered ones back into the thread, then set status back to awaiting_pm. The PM replies again, another webhook fires, we loop back into this same function.
Step 22 — Outcome C: All resolved → draft
clarifications_md = _build_clarifications_md(ctx, sc_id)
messages = build_draft_messages(story, clarifications_md, session["hits"])
response, _ = call_chat(messages, model=config.CHAT_MODEL)
draft = parse_draft_result(response)
db.update_draft(sc_id, draft)
db.update_status(sc_id, "done", "drafting")
shortcut_client.post_comment(sc_id, f"02_patch.sql\n...\n03_revert.sql\n...", parent_id=thread_comment_id)
Everything is known. _build_clarifications_md converts the SessionContext into a structured markdown string — template, confirmed values, resolved questions, extraction summary. That gets passed to the Sonnet draft call.
build_draft_messages builds the prompt the same way build_clarify_messages did — system message = draft.md + rules + templates + glossary, user message = story + clarifications_md + past patch SQL from hits.
parse_draft_result parses the response into a DraftResult with patch_sql, revert_sql, rationale. Saved to the DB, status set to done, scripts posted into the thread. Session is finished.
The state machine in full
Engineer: "@agentbot clarify"
│
_run_clarify() ← Sonnet, posts 01_extract.sql
│
awaiting_extraction
│
(GitLab CI submits (PM replies in thread)
via /submitextraction) │
│ │
└──────────┬───────────────┘
│
_preprocess_and_advance() ← Haiku
│
┌──────────┼──────────┐
reclarify? still_open? all resolved
│ │ │
awaiting_ awaiting_ done
extraction pm (Sonnet, posts
(Sonnet, (loop back 02_patch.sql +
round 2) on reply) 03_revert.sql)
Ready to move to submit_extraction next?