Retrieval and memory¶

A good answer comes from your help center, not from the model's memory. But retrieved text is risky. It can belong to another customer, it can carry injected instructions, and the model can claim it cited something it never saw. The harness governs retrieval: it re-checks every item's access list, labels it untrusted, caps how much reaches the model, and runs your context guards, before anything gets to the model.

Some help articles¶

Add data/help.yaml. The third article is for one customer only:

data/help.yaml

documents:
  - id: help-refunds
    content: A refund takes up to five business days to reach the card you paid with.
  - id: help-invoices
    content: Every invoice is under Settings, Billing, where you can download it as a PDF.
  - id: help-globex-sla
    content: Globex's contract promises a reply from support within four hours.
    acl: { tenants: [globex] }

acl: { tenants: [globex] } says only runs for the tenant globex may see that article. An item can also list principals and roles. Empty lists restrict nothing.

Declare a context source¶

harness.yaml (the context section)

context:
  help:
    use: memory
    file: data/help.yaml
    max_items: 3

help is a source that uses the built-in memory provider: documents held in memory, from the file, searched lexically. max_items keeps the best three.

Fetch, then answer with citations¶

helpdesk/flows.py

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

import harnessweaver as hw

OUTAGE_REPLY = "We know about the outage and are fixing it. Updates: status.acme.example"


async def answer(s: hw.Session, question: str) -> str:
    """Answer a customer's question from the help center."""
    route = await s.decide("intent", text=question)
    if route.choice == "outage":
        return OUTAGE_REPLY
    found = await s.context("help").fetch(question)
    articles = "\n".join(f"[{item.id}] {item.content}" for item in found)
    return await s.generate(prompt="answer@2", question=question, articles=articles)

The prompt lists the articles by id and asks for a citation:

prompts/answer@2.md

You answer questions for Acme's support desk, in two sentences or fewer.
Use only these help articles, and cite the one you used in square brackets,
like [help-refunds].

{articles}

Question: {question}

Then make the harness check the citations, with a guard on the answer task's output only:

harness.yaml (the answer task)

  answer:
    runner: { use: session, handler: "helpdesk.flows:answer" }
    output: str
    budget: { max_model_calls: 2, max_usd: 0.02 }
    guardrails:
      output:
        - { id: cited, use: grounded.citations, action: repair }

grounded.citations checks that the answer's sentences cite an item in square brackets, such as [help-refunds], and that each citation names an item this run actually retrieved. A made-up source doesn't count. action: repair sends an uncited answer back to the model. It's a task guard because the triage task's output isn't prose.

Run it¶

harness.offline.yaml

models:
  provider: scripted
  rules:
    - contains: "Question:"
      reply: { text: "A refund takes up to five business days to reach your card [help-refunds]." }
  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
A refund takes up to five business days to reach your card [help-refunds].

Who sees what¶

Fetch the same question for two tenants:

main_fetch.py

import asyncio

import harnessweaver as hw


async def main() -> None:
    harness = hw.load("harness.yaml")
    question = "How fast does support reply to a refund question?"
    for tenant in ("acme", "globex"):
        async with harness.session("answer", principal="ann", tenant=tenant) as s:
            items = await s.context("help").fetch(question)
            print(tenant, [item.id for item in items])


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

$ HARNESSWEAVER_ENV=offline uv run python main_fetch.py
acme ['help-refunds', 'help-invoices']
globex ['help-refunds', 'help-globex-sla', 'help-invoices']

Acme never sees Globex's contract terms. The provider returned the article, and the harness dropped it, because its ACL excludes the run's tenant.

How retrieval is governed¶

Every item, fetched or passed in, is governed inside the context call, before any context.after middleware sees it, so no middleware order can let a bad item through:

ACL re-check. An item whose ACL excludes the run's tenant or principal is dropped, even when its provider returned it. With require_acl: true, an item with no ACL is dropped too.
Labels. The item's source becomes the source's name, and it is untrusted unless the source is label: trusted (RULES S3).
Caps. max_items, then max_tokens, in order.
Each drop is a harness.context note in the trace, with the item's id and never its content.

Then the context guard stage runs on each kept item. There a guard's block drops the item, not the run:

guardrails:
  context:
    - { id: context-injection, use: injection.heuristic, action: block }

Untrusted items that are kept mark the run as having seen untrusted content. That label matters in Tools: a run that has seen untrusted content and private data can't send data out without a person's approval.

Your own search¶

The built-in providers are memory (documents in memory, searched lexically) and passthrough, which fetches nothing. Vector stores and search services come as plugins. Until you have one, retrieve with your own client and let the harness govern the results:

context:
  web: { use: passthrough }

items = [hw.ContextItem(id=hit.id, content=hit.text, source="web") for hit in my_search(question)]
checked = await s.context("web").check(items)

The same ACL, label, cap and guard rules apply. Providers lists every source setting.

Memory¶

s.memory is a small key-value store for the run's tenant, principal and task:

await s.memory.put("last_order", "A-1002", ttl=timedelta(hours=1))
order = await s.memory.get("last_order")   # None once expired or absent

The harness picks the namespace, so no tenant or principal reads another's entries. Entries expire after ttl, or the spec's memory.default_ttl. It is a store, not a memory model: what an agent remembers, and when, is your app's business (RULES R12).

Recap¶

A context source fetches items. The harness re-checks ACLs, labels items untrusted, caps them, and runs the context guards before the model sees anything.
grounded.citations makes an answer cite only what the run really retrieved.
s.memory keeps namespaced key-value entries per tenant, principal and task.