Runnable correction fixtures: automation-reference-examples.

Follow @renezander030 for tested implementation notes.

Use the scoped implementation and version described below. Inspect this reference and its retained source filename before copying it:

gh gist view 4c97bc473db055f2aa6d71be4c4551ce --files

This command lists the public reference files; it does not exercise your live environment. The existing snippets and tests below define the actual setup contract.

A context layer over chat, tickets, wiki, CRM, mailboxes and meeting notes has to answer as the person asking. The usual build gets this wrong in a way that passes every demo: one service account reads everything, an index is built from it, and a filter at query time trims the result. The filter runs after the read. It leaks in 3 places.

Recommended: recheck current source permissions before loading/rendering results and every required provenance item. Filter first, then compute visible counts. Omit denied items entirely; never emit their titles, hidden total, IDs or provenance hints. Authorization and loading need a provider snapshot/transaction when eliminating a permission-change race is part of the contract.

git clone https://github.com/renezander030/automation-reference-examples.git
cd automation-reference-examples
export PYTHONPATH="$PWD/python"
python3 -m unittest discover -s tests -v

from acl import retrieve
records = {'a': {'title': 'Visible', 'sources': ['source']}}
grants = {'a', 'source'}
authorize = lambda principal, item: item in grants
print(retrieve(['a'], 'reader', authorize, records.__getitem__))
grants.remove('source')
print(retrieve(['a'], 'reader', authorize, records.__getitem__))

Pass: first visible count is 1; after revocation it is 0, with no title or source leakage. Fixture source. This tests request-time checks and provenance omission, not an atomic ACL snapshot on a remote SaaS API.

Last tested: October 2026 — synthetic permission/revocation fixture only.

1. Counts. "How many pages in the sales space mention Acme" answered with a number, for a caller who cannot open the sales space.
2. Titles. A trimmed result that still shows the title of a meeting the caller was not in.
3. Provenance chains. A fact shared with the caller whose chain links to a write-up the caller cannot open, so the chain becomes a directory of things they are not allowed to see.

• Every read runs delegated, with the caller's own token. The layer holds no permissions of its own, the sources enforce theirs, which they already do well.
• Group membership is resolved from the directory at question time, so a removal takes effect on the next question, with no re-index.
• Cite the passage that put an item in front of the caller, or drop the item. Never say how many were dropped.
• A chain shows only records the caller can open. A closed hop is dropped, not counted, not titled.
• Cache authorization by identity and a current permission generation, or revalidate permissions on every request. Identity-only caches can retain revoked access.

from dataclasses import dataclass

@dataclass
class Hit:
    system: str
    record_id: str
    passage: str      # the verbatim mention that put this item in front of the caller
    url: str

class Source:
    def as_user(self, caller):
        """Return a client bound to the caller's delegated token. Never a service account."""
        raise NotImplementedError

def retrieve(question, caller, sources):
    items = []
    for source in sources:
        client = source.as_user(caller)          # the source applies its own ACL
        for hit in client.search(question):
            if not hit.passage:                  # cite the mention or drop the item
                continue
            items.append(hit)
    return items                                 # no total, no "n more you cannot see"

def answer(question, caller, sources):
    items = retrieve(question, caller, sources)
    if not items:
        return {"found": False, "items": []}    # an explicit not found, never a count of the unseen
    return {"found": True, "items": [i.__dict__ for i in items]}

def chain_for(fact, caller, sources):
    links = []
    for ref in fact.provenance:                                        # record ids only, never titles
        rec = sources[ref.system].as_user(caller).get(ref.record_id)   # None when the caller cannot open it
        if rec is None:
            continue                                                   # a closed hop is dropped, not counted, not titled
        links.append({"system": ref.system, "title": rec.title, "url": rec.url})
    return links

• count_never_includes_unopenable: a caller without access to space X asks a count question; the answer contains no number derived from X, and it is an explicit not found when nothing else matches.
• title_never_leaks_via_result_or_chain: a fact whose source record is unreadable for the caller renders with the fact and without the source title.
• cache_is_per_identity: 2 callers, same question, no shared cache entry between them.
• membership_resolved_at_question_time: remove the caller from a group; the next question no longer returns that group's records, with no re-index in between.

This is a retrieval-boundary design. The executable public fixture below uses synthetic records; it does not claim to reproduce a private tenant deployment.

The write side, proposals a person decides and one writer for the graph, is #18: https://gist.github.com/renezander030/a3600b3378b01f00080f9f41150f16d2

• Permission revocation and executable fixtures; historical tests are not refreshed.

Companion reference for Production AI Automation Notes: #1 index and #15 approval evidence. No new series slot is assigned.

Follow @renezander030 for tested implementation notes.

Comment with runtime/version and redacted configuration, the failing input and what passed or failed. Include machine/RAM/backend/model/quant for local inference.