Your own steps: sessions¶

The direct runner does one thing: send the input to the model and return the answer. Many tasks are more than that: decide what kind of question this is, fetch some documents, then ask the model. In a session, your code owns those steps, and the harness governs each one: every model call, tool call and retrieval passes the same hooks, guards and budgets as a runner's.

This chapter adds a second task, answer, written as a session.

Write the steps¶

Add helpdesk/flows.py:

helpdesk/flows.py

"""The answer task: our own steps, each one governed by the harness."""

import harnessweaver as hw


async def answer(s: hw.Session, question: str) -> str:
    """Answer a customer's question in the support desk's voice."""
    return await s.generate(prompt="answer@1", question=question)

s is a Session. s.generate renders the prompt answer@1 with question, sends it to the model as a user message, and returns the answer. The model call is governed exactly as a task's is. Add the prompt:

prompts/answer@1.md

You answer questions for Acme's support desk, in two sentences or fewer.

Question: {question}

Run the steps as a task¶

Add the answer task, with the session runner pointing at your function:

harness.yaml (tasks and validation)

tasks:
  triage:
    runner: direct
    model: fast
    prompt: triage@2
    input: helpdesk.models:Ticket
    output: helpdesk.models:Triage
    budget:
      max_model_calls: 3
      max_usd: 0.01
      deadline: 30s
      overridable: [max_model_calls, max_usd]
  answer:
    runner: { use: session, handler: "helpdesk.flows:answer" }
    output: str
    budget: { max_model_calls: 2, max_usd: 0.02 }

validation:
  rules:
    - id: bugs-are-urgent
      when: "task.name == 'triage' && output.category == 'bug' && output.priority > 2"
      message: "a bug is priority 1 or 2"
    - id: short-summary
      when: "task.name == 'triage' && size(output.summary) > 120"
      message: "keep the summary under 120 characters"

Notice the rules changed too. Rules check every task's output, and answer returns plain text, which has no category or summary. A rule that can't be evaluated counts as broken, so without task.name == 'triage' every answer would be sent back for repair.

Give the offline model an answer for questions:

harness.offline.yaml

models:
  provider: scripted
  rules:
    - contains: "Question:"
      reply: { text: "Refunds go back to the card you paid with within five business days." }
  replies:
    - output:
        category: billing
        priority: 2
        summary: Charged twice for the Pro plan and wants one payment refunded.

main.py

import asyncio

import harnessweaver as hw


async def main() -> None:
    harness = hw.load("harness.yaml")
    question = "How long does a refund take?"
    result = await harness.run("answer", question, principal="ann", tenant="acme")
    print(result.status.value)
    print(result.output)


if __name__ == "__main__":
    asyncio.run(main())

$ HARNESSWEAVER_ENV=offline uv run python main.py
completed
Refunds go back to the card you paid with within five business days.

As a task, answer gets everything a task has: the input guards run on the question before your function sees it, the task's budget applies, whatever the function returns is checked by the output guards and validation rules, and the run can be retried by idempotency key, suspended for approval and resumed, and graded by evals.

principal and tenant say who the run acts for. Policy rules, retrieval ACLs and rate limits all read them, as later chapters show.

Or open a session yourself¶

In an app that already has its own request handling, you can open a session around your code instead of defining a task's runner:

main_session.py

import asyncio

from helpdesk.flows import answer

import harnessweaver as hw


async def main() -> None:
    harness = hw.load("harness.yaml")
    async with harness.session("answer", principal="ann", tenant="acme") as s:
        question = await s.guard("input", "How long does a refund take? I'm ann@example.com")
        print(question)
        print(await answer(s, str(question)))


if __name__ == "__main__":
    asyncio.run(main())

$ HARNESSWEAVER_ENV=offline uv run python main_session.py
How long does a refund take? I'm [EMAIL]
Refunds go back to the card you paid with within five business days.

harness.session("answer", ...) opens one run with the answer task's budget and controls. Nobody hands a standalone session its input, so the input guards run when your code asks: s.guard("input", ...) returned the question with the e-mail address redacted. When the async with block ends, run.end fires and the run's result is kept.

What a session can do¶

Method	What it does	Hook point
`s.guard("input", content)`	Runs the input guards; returns the content, possibly redacted	`input.before`
`s.decide(router, text=...)`	Asks a router to choose (Routing)	`route.decide`
`s.context(source).fetch(query)`	Retrieves items, governed (Retrieval)	`context.after`
`s.context(source).check(items)`	Governs items your own code retrieved	`context.after`
`s.generate(prompt=..., output=..., model=..., **variables)`	A governed model call; `output` parses into a type and repairs	`model.before`, `model.after`, `output.before`
`s.tool(name, **arguments)`	A governed tool call; returns its output (Tools)	`tool.before`, `tool.after`
`s.validate("output", value, output=...)`	Checks an output you produced yourself	`output.before`
`s.prompt(ref, **variables)`	Renders a versioned prompt without calling a model	-
`s.memory`	Key-value entries for this tenant, principal and task (Retrieval and memory)	-
`s.run_id`, `s.envelope`, `s.trace`	The run's id, its envelope and its trace so far	-

A control that stops the run, such as a guard that blocks or a budget that runs out, raises a RunInterrupted error in your code: RunBlocked or RunEscalated. Your code doesn't need to catch it, because the run is over either way.

Task or standalone?

A standalone session can't be suspended and resumed later, because the harness can't re-enter your code. So in a standalone session, a request for approval escalates instead. As a task (runner: session), the same function suspends, and after an approval it runs again from the start. The approved call then passes without asking again.

Three ways to adopt HarnessWeaver¶

You have now used two of the three integration styles:

Style	You write	The harness guarantees
Run (`harness.run` with `direct`)	A task in the spec	Every hook, plus idempotency, approvals that suspend and resume, and repairs
Session (`runner: session` or `harness.session`)	The steps, in your code	Every hook your code asks for
Wrap (`harness.instrument(client)`)	Nothing new: your existing client	Every call it makes is traced, costed, budgeted and guarded

Integration styles compares them, and Govern an existing client shows the third.

Recap¶

A session runs your code's steps, each governed: s.generate, s.tool, s.context, s.decide, s.guard.
runner: { use: session, handler: ... } makes a function a task. harness.session(...) opens one around existing code.
Rules check every task's output, so scope them with task.name.