diff --git a/src/openultrasast/semantic/obligations/operations.py b/src/openultrasast/semantic/obligations/operations.py
new file mode 100644
index 0000000..cd88ea7
--- /dev/null
+++ b/src/openultrasast/semantic/obligations/operations.py
@@ -0,0 +1,248 @@
+"""Obligated operations and discharge witnesses in one parsed file (authorization-obligations, Req 3.1, 5.4).
+
+An ``Operation`` is a call site matching an operation fact inside a named function. A ``Discharge`` is a witness that a
+discharger fact is present: a decorator on the handler, a guard or check call before the operation, an identity
+constraint keyword bound from the authenticated context, or a non-permissive setting. ``covers`` says whether one witness
+discharges one operation: same function, a kind the operation requires, and, for identity constraints, a value bound from
+the authenticated context (a constraint bound from the request or a constant discharges nothing).
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from ..facts import SemanticFacts
+from ..ir import FileIR, FunctionIR
+from ..variants import trailing_name
+from .facts import DischargerFact, ObligationFacts
+from .shapes import _call_matches, _constraint_provenance, _is_literal, _operations, _permissive
+
+SCOPES = ("decorator", "router", "statement", "hop")
+
+
+@dataclass(frozen=True)
+class Operation:
+    path: str
+    line: int
+    function: str
+    kind: str
+    fact_id: str
+    resource: str | None  # identifier token naming the resource (model or table), never text
+    requires: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class Discharge:
+    path: str
+    line: int | None
+    function: str
+    kind: str
+    fact_id: str
+    provenance: str
+    scope: str  # SCOPES
+    detail: str = ""  # identifier only: the decorator or call name that witnessed the discharge
+
+
+def find_operations(ir: FileIR, facts: ObligationFacts, *, text: str) -> tuple[Operation, ...]:
+    """Every obligated operation in the file's named functions, in source order."""
+    if not ir.parse_ok:
+        return ()
+    lines = text.splitlines()
+    found: list[Operation] = []
+    seen: set[tuple[str, int, str]] = set()
+    for function in ir.functions:
+        if function.name == "<module>":
+            continue
+        for site in _operations(function, facts, lines, kind=None):
+            key = (function.name, site.call.line, site.fact.id)
+            if key in seen:
+                continue  # the walker records a chained `x.filter_by(...).first()` as two call sites on one line
+            seen.add(key)
+            found.append(
+                Operation(
+                    path=ir.path,
+                    line=site.call.line,
+                    function=function.name,
+                    kind=site.fact.kind,
+                    fact_id=site.fact.id,
+                    resource=_resource_token(site.call.name, site.line_text, site.fact.resource_arg, site.call.arg_texts),
+                    requires=site.fact.requires,
+                )
+            )
+    return tuple(found)
+
+
+def find_discharges(
+    ir: FileIR,
+    facts: ObligationFacts,
+    flow_facts: SemanticFacts,
+    *,
+    text: str,
+    entries: Sequence[object] = (),
+) -> tuple[Discharge, ...]:
+    """Every discharge witness in the file: decorators above each handler, guard and check calls, identity constraints."""
+    if not ir.parse_ok:
+        return ()
+    lines = text.splitlines()
+    evidence_by_function: dict[str, list[str]] = {}
+    for entry in entries:
+        name = getattr(entry, "function_name", None)
+        if name:
+            evidence_by_function.setdefault(str(name), []).extend(str(item) for item in getattr(entry, "access_evidence", []) or [])
+    witnesses: list[Discharge] = []
+    for function in ir.functions:
+        if function.name == "<module>":
+            continue
+        decorators = _decorators_above(function, lines) + evidence_by_function.get(function.name, [])
+        for fact in facts.dischargers:
+            witnesses.extend(_decorator_witnesses(ir.path, function, fact, decorators))
+            witnesses.extend(_call_witnesses(ir.path, function, fact))
+            witnesses.extend(_constraint_witnesses(ir.path, function, fact, facts, flow_facts, lines))
+    return tuple(_dedupe(witnesses))
+
+
+def covers(witness: Discharge, operation: Operation) -> bool:
+    """A witness discharges an operation when it sits in the same function, has a kind the operation requires, and, for an
+    identity constraint, binds its value from the authenticated context. Ordering along the path is the Dominance protocol's
+    job (task 2.4); this is the kind-and-provenance fit only."""
+    if witness.path != operation.path or witness.function != operation.function:
+        return False
+    if witness.kind not in operation.requires:
+        return False
+    if witness.kind == "identity_constraint" and witness.provenance != "authenticated_context":
+        return False
+    return not (witness.kind == "non_permissive_value" and witness.line != operation.line)
+
+
+def _decorators_above(function: FunctionIR, lines: Sequence[str]) -> list[str]:
+    start = function.start_line
+    found: list[str] = []
+    index = start - 2
+    while index >= 0 and lines[index].strip().startswith("@"):
+        found.append(lines[index].strip())
+        index -= 1
+    return found
+
+
+def _decorator_witnesses(path: str, function: FunctionIR, fact: DischargerFact, decorators: Sequence[str]) -> list[Discharge]:
+    out: list[Discharge] = []
+    for decorator in decorators:
+        name = _decorator_name(decorator)
+        if any(candidate == name or candidate in decorator for candidate in fact.decorators):
+            out.append(
+                Discharge(
+                    path=path,
+                    line=None,
+                    function=function.name,
+                    kind=fact.kind,
+                    fact_id=fact.id,
+                    provenance="authenticated_context" if fact.kind == "path_guard" else "unknown",
+                    scope="decorator",
+                    detail=name,
+                )
+            )
+    return out
+
+
+def _call_witnesses(path: str, function: FunctionIR, fact: DischargerFact) -> list[Discharge]:
+    if fact.kind == "identity_constraint":
+        return []
+    out: list[Discharge] = []
+    for call in function.calls:
+        if any(_call_matches(call.name, name) for name in fact.calls):
+            if fact.kind == "non_permissive_value":
+                continue  # settings are judged per operation line in _constraint_witnesses
+            out.append(
+                Discharge(
+                    path=path,
+                    line=call.line,
+                    function=function.name,
+                    kind=fact.kind,
+                    fact_id=fact.id,
+                    provenance="authenticated_context" if fact.kind == "path_guard" else "unknown",
+                    scope="statement",
+                    detail=trailing_name(call.name),
+                )
+            )
+    return out
+
+
+def _constraint_witnesses(
+    path: str,
+    function: FunctionIR,
+    fact: DischargerFact,
+    facts: ObligationFacts,
+    flow_facts: SemanticFacts,
+    lines: Sequence[str],
+) -> list[Discharge]:
+    out: list[Discharge] = []
+    if fact.kind == "identity_constraint":
+        for site in _operations(function, facts, lines, kind=None):
+            provenance = _constraint_provenance(site, fact, flow_facts)
+            if provenance is not None:
+                out.append(
+                    Discharge(
+                        path=path,
+                        line=site.call.line,
+                        function=function.name,
+                        kind="identity_constraint",
+                        fact_id=fact.id,
+                        provenance=provenance,
+                        scope="statement",
+                        detail=trailing_name(site.call.name),
+                    )
+                )
+    elif fact.kind == "non_permissive_value":
+        for site in _operations(function, facts, lines, kind="security_setting"):
+            if any(_call_matches(site.call.name, name) for name in fact.calls) and not _permissive(site, fact):
+                out.append(
+                    Discharge(
+                        path=path,
+                        line=site.call.line,
+                        function=function.name,
+                        kind="non_permissive_value",
+                        fact_id=fact.id,
+                        provenance="constant",
+                        scope="statement",
+                        detail=trailing_name(site.call.name),
+                    )
+                )
+    return out
+
+
+def _resource_token(callee: str, line_text: str, resource_arg: int | None, arg_texts: Sequence[str]) -> str | None:
+    """The identifier that names the resource: the receiver's first segment (`Book.query...` -> `book`), a table name in a
+    SQL literal, or the argument the fact points at. Never more than one identifier."""
+    if resource_arg is not None and resource_arg < len(arg_texts):
+        match = re.match(r"[A-Za-z_][\w]*", arg_texts[resource_arg].strip())
+        return match.group(0).lower() if match else None
+    head = callee.split(".")[0].split("(")[0]
+    if (
+        head
+        and head not in {"self", "cls", "db", "session", "cursor", "conn", "connection", "c", "cur", "request", "req", "this"}
+        and not _is_literal(head)
+    ):
+        return head.lower()
+    table = re.search(r"\b(?:from|into|update|join)\s+([A-Za-z_][\w]*)", line_text, re.I)
+    return table.group(1).lower() if table else None
+
+
+def _decorator_name(decorator: str) -> str:
+    body = decorator.lstrip("@").strip()
+    return trailing_name(body.split("(", 1)[0])
+
+
+def _dedupe(witnesses: Sequence[Discharge]) -> list[Discharge]:
+    seen: set[tuple[object, ...]] = set()
+    out: list[Discharge] = []
+    for witness in witnesses:
+        key = (witness.path, witness.line, witness.function, witness.kind, witness.provenance, witness.scope, witness.detail)
+        if key not in seen:
+            seen.add(key)
+            out.append(witness)
+    return out
+
+
+__all__ = ["SCOPES", "Discharge", "Operation", "covers", "find_discharges", "find_operations"]
diff --git a/tests/test_obligation_operations.py b/tests/test_obligation_operations.py
new file mode 100644
index 0000000..3680a54
--- /dev/null
+++ b/tests/test_obligation_operations.py
@@ -0,0 +1,92 @@
+"""authorization-obligations task 2.1: operations and discharge witnesses in a parsed file (offline, stdlib ast)."""
+
+from __future__ import annotations
+
+from openultrasast.mapping import EntryPointRecord
+from openultrasast.semantic.facts import load_facts
+from openultrasast.semantic.ir import parse_file
+
+APP = (
+    "from flask import request\n\n\n"
+    "@app.route('/books/<title>')\n@login_required\ndef guarded(title):\n"
+    "    return Book.query.filter_by(book_title=title).first()\n\n\n"
+    "@app.route('/books/mine/<title>')\ndef constrained(title):\n"
+    "    user_id = request.user.id\n"
+    "    return Book.query.filter_by(user_id=user_id, book_title=title).first()\n\n\n"
+    "@app.route('/books/any/<title>')\ndef leaky(title):\n"
+    "    owner = request.args.get('owner')\n"
+    "    return Book.query.filter_by(user_id=owner, book_title=title).first()\n\n\n"
+    "@app.route('/health')\ndef health():\n    return 'ok'\n"
+)
+
+
+def _entry(function: str, line: int, end: int, access: str, evidence: list[str]) -> EntryPointRecord:
+    return EntryPointRecord(
+        path="app.py",
+        line=line,
+        end_line=end,
+        function_name=function,
+        name=function,
+        kind="route",
+        access_level=access,  # type: ignore[arg-type]
+        trust_boundary="http_request",
+        access_evidence=evidence,
+        conditions=[],
+        provenance="test",
+        rationale="",
+    )
+
+
+ENTRIES = [
+    _entry("guarded", 6, 7, "authenticated", ["@login_required"]),
+    _entry("constrained", 11, 13, "public", []),
+    _entry("leaky", 17, 19, "public", []),
+    _entry("health", 23, 24, "public", []),
+]
+
+
+def test_find_operations_names_kind_fact_function_and_resource() -> None:
+    from openultrasast.semantic.obligations import load_obligation_facts
+    from openultrasast.semantic.obligations.operations import find_operations
+
+    ir = parse_file("app.py", APP, "python")
+    ops = find_operations(ir, load_obligation_facts().for_language("python"), text=APP)
+    assert [(op.function, op.kind, op.resource) for op in ops] == [
+        ("guarded", "protected_read", "book"),
+        ("constrained", "protected_read", "book"),
+        ("leaky", "protected_read", "book"),
+    ]
+    assert all(op.path == "app.py" and op.fact_id == "orm-read" and op.line > 0 for op in ops)
+    assert not [op for op in ops if op.function == "health"]  # a constant return is no operation
+
+
+def test_find_discharges_reports_witness_kind_provenance_and_scope() -> None:
+    from openultrasast.semantic.obligations import load_obligation_facts
+    from openultrasast.semantic.obligations.operations import find_discharges
+
+    ir = parse_file("app.py", APP, "python")
+    witnesses = find_discharges(
+        ir, load_obligation_facts().for_language("python"), load_facts().for_language("python"), text=APP, entries=ENTRIES
+    )
+    by_function = {}
+    for w in witnesses:
+        by_function.setdefault(w.function, []).append((w.kind, w.provenance, w.scope))
+    assert ("path_guard", "authenticated_context", "decorator") in by_function["guarded"]
+    assert ("identity_constraint", "authenticated_context", "statement") in by_function["constrained"]
+    assert ("identity_constraint", "request_input", "statement") in by_function["leaky"]  # present, but discharges nothing
+    assert "health" not in by_function
+    assert all(w.path == "app.py" and w.fact_id for w in witnesses)
+
+
+def test_discharge_witness_covers_operation_only_when_kind_and_provenance_fit() -> None:
+    from openultrasast.semantic.obligations import load_obligation_facts
+    from openultrasast.semantic.obligations.operations import covers, find_discharges, find_operations
+
+    ir = parse_file("app.py", APP, "python")
+    facts = load_obligation_facts().for_language("python")
+    ops = {op.function: op for op in find_operations(ir, facts, text=APP)}
+    witnesses = find_discharges(ir, facts, load_facts().for_language("python"), text=APP, entries=ENTRIES)
+    covering = {op.function: [w.kind for w in witnesses if covers(w, op)] for op in ops.values()}
+    assert covering["constrained"] == ["identity_constraint"]
+    assert covering["leaky"] == []  # request-bound identity does not cover
+    assert covering["guarded"] == []  # a path guard does not cover a protected read (only privileged actions)