INFD-WP-0010: present_existing intake profile (S1) and owner answer to decision services alignment

A registered submitter with a present_existing profile may pose a memo that
pins an approval it created itself in approval-engine. Intake verifies the
object read-only (requested, no entries, human_control, actor == profile
actor, digest == memo, binding == intent, required_count 1, unexpired, one
memo per approval) and never creates, enters or consumes. The recorded
requester reaches review and the access-engine context. Existing submitters
are unchanged.

Answer doc accepts S1 (profile, not route; packet carries the admission),
S4 in principle (waits on gate-house classes and policy) and S3 (exists
behind INFD_HUB_WRITEBACK). Workplan INFD-WP-0010 proposed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 5502@bnt-lap001
Assistant-Session: 68b3b02f-5d0f-4e71-8421-796f62e50196

informed_decision/intake.py

@@ -41,6 +41,7 @@ _MEMO_ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]{2,62}$")
 _SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{1,62}$")
 _UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
 _SHA_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
+_SLUG_KEY_RE = re.compile(r"^[a-z][a-z0-9_]{0,62}$")
 
 #: Secret markers (§3.3.6). Shapes, not values; deliberately broad.
 SECRET_MARKERS: tuple[tuple[str, re.Pattern[str]], ...] = (
@@ -74,6 +75,18 @@ class SubmitterGrant:
     review_groups: frozenset[str]
 
 
+@dataclass(frozen=True)
+class PresentExisting:
+    """Submitter profile ``present_existing`` (INFD-WP-0010-T01).
+
+    The submitter's approval already exists in approval-engine, created by its
+    own actor. Intake only verifies and presents it; it never creates one.
+    """
+
+    approval_actor: str
+    actions: frozenset[str]
+
+
 @dataclass(frozen=True)
 class ValidatedSubmission:
     submitter_id: str
@@ -86,6 +99,7 @@ class ValidatedSubmission:
     approval_binding: dict
     memo: Memo
     packet: dict[str, bytes]
+    profile: str = "create"
 
 
 def _closed(value: Any, keys: set[str], required: set[str], path: str) -> dict:
@@ -155,16 +169,21 @@ def idempotency_outcome(stored_digest: str | None, digest: str) -> str:
     raise IntakeRefused("idempotency_conflict", "$.idempotency_key")
 
 
-def _check_memo(memo: Any, packet_hashes: dict[str, str]) -> Memo:
+def _check_memo(memo: Any, packet_hashes: dict[str, str], *, pinned: bool = False) -> Memo:
     memo = _closed(memo, MEMO_KEYS, MEMO_KEYS - {"locale", "ui_release"}, "$.memo")
     _text(memo["id"], "$.memo.id", max_len=63, pattern=_MEMO_ID_RE)
     if memo["version"] != 1 or isinstance(memo["version"], bool):
         raise IntakeRefused("draft_version_must_be_1", "$.memo.version")
     if memo["sealed"] is not False:
         raise IntakeRefused("draft_must_be_unsealed", "$.memo.sealed")
-    for field in ("approval_id", "approval_binding_digest"):
-        if memo[field] is not None:
-            raise IntakeRefused("approval_pinned_by_submitter", f"$.memo.{field}")
+    if pinned:
+        # present_existing: the memo must name the existing approval exactly.
+        _text(memo["approval_id"], "$.memo.approval_id", max_len=36, pattern=_UUID_RE)
+        _text(memo["approval_binding_digest"], "$.memo.approval_binding_digest", max_len=71, pattern=_SHA_RE)
+    else:
+        for field in ("approval_id", "approval_binding_digest"):
+            if memo[field] is not None:
+                raise IntakeRefused("approval_pinned_by_submitter", f"$.memo.{field}")
     check_one_question(_text(memo["question"], "$.memo.question"))
     _text(memo["requested_act"], "$.memo.requested_act", max_len=40, pattern=_SLUG_RE)
     _text(memo["brief"], "$.memo.brief", max_len=20000)
@@ -220,8 +239,26 @@ def _check_memo(memo: Any, packet_hashes: dict[str, str]) -> Memo:
         raise IntakeRefused("memo_unparseable", "$.memo") from None
 
 
-def validate_submission(envelope: Any, *, submitter_id: str, grant: SubmitterGrant | None) -> ValidatedSubmission:
-    """Validate one intake envelope for an already-authenticated submitter."""
+def _existing_target(value: Any, path: str) -> dict:
+    """An existing approval's target may have any flat shape (BEA uses admission id/digest)."""
+    if not isinstance(value, dict) or not 1 <= len(value) <= 8:
+        raise IntakeRefused("invalid_target", path)
+    for key, item in value.items():
+        if not isinstance(key, str) or not _SLUG_KEY_RE.match(key):
+            raise IntakeRefused("invalid_target", path)
+        _text(item, f"{path}.{key}", max_len=200)
+    return dict(value)
+
+
+def validate_submission(envelope: Any, *, submitter_id: str, grant: SubmitterGrant | None,
+                        present_existing: PresentExisting | None = None) -> ValidatedSubmission:
+    """Validate one intake envelope for an already-authenticated submitter.
+
+    With a ``present_existing`` profile the memo must pin an existing approval
+    and the intent names that approval's binding; the actor is the profile's,
+    never the body's. Without it, behaviour is unchanged: the memo must not pin
+    an approval and intake will create one as ``informed-decision``.
+    """
     if grant is None or grant.submitter_id != submitter_id:
         raise IntakeRefused("submitter_unknown", "$")
     env = _closed(envelope, ENVELOPE_KEYS, ENVELOPE_REQUIRED, "$")
@@ -247,9 +284,14 @@ def validate_submission(envelope: Any, *, submitter_id: str, grant: SubmitterGra
     intent = _closed(env["approval_intent"], INTENT_KEYS, INTENT_KEYS, "$.approval_intent")
     for field in ("action", "principal", "purpose"):
         _text(intent[field], f"$.approval_intent.{field}", max_len=400)
-    target = _closed(intent["target"], TARGET_KEYS, TARGET_KEYS, "$.approval_intent.target")
-    for field in TARGET_KEYS:
-        _text(target[field], f"$.approval_intent.target.{field}", max_len=200)
+    if present_existing is not None:
+        target = _existing_target(intent["target"], "$.approval_intent.target")
+        if intent["action"] not in present_existing.actions:
+            raise IntakeRefused("action_not_admitted", "$.approval_intent.action")
+    else:
+        target = _closed(intent["target"], TARGET_KEYS, TARGET_KEYS, "$.approval_intent.target")
+        for field in TARGET_KEYS:
+            _text(target[field], f"$.approval_intent.target.{field}", max_len=200)
 
     packet = env["packet"]
     if not isinstance(packet, list) or not packet:
@@ -270,7 +312,7 @@ def validate_submission(envelope: Any, *, submitter_id: str, grant: SubmitterGra
         contents[item["item_id"]] = data
         hashes[item["item_id"]] = "sha256:" + hashlib.sha256(data).hexdigest()
 
-    memo = _check_memo(env["memo"], hashes)
+    memo = _check_memo(env["memo"], hashes, pinned=present_existing is not None)
 
     if memo_class not in grant.memo_classes:
         raise IntakeRefused("memo_class_not_admitted", "$.memo_class")
@@ -286,7 +328,9 @@ def validate_submission(envelope: Any, *, submitter_id: str, grant: SubmitterGra
         drafted_by=drafted_by,
         serves=dict(serves) if serves else None,
         approval_binding={**{k: intent[k] for k in ("action", "principal", "purpose")},
-                          "actor": REQUESTER_ACTOR, "target": dict(target)},
+                          "actor": present_existing.approval_actor if present_existing else REQUESTER_ACTOR,
+                          "target": dict(target)},
         memo=memo,
         packet=contents,
+        profile="present_existing" if present_existing else "create",
     )

informed_decision/intake_approval.py

@@ -1,5 +1,9 @@
-"""Create-only approval adapter; token custody/exchange belongs to workload setup."""
+"""Create-only approval adapter and the present_existing read-only adapter.
+
+Token custody/exchange belongs to workload setup.
+"""
 from datetime import datetime, timedelta, timezone
+import re
 
 from .http_transport import JSONTransport, fixed_origin
 from .intake import canonical
@@ -32,3 +36,34 @@ class IntakeApprovalClient:
             return result
         except Exception:
             raise IntakeError(503, 'approval_create_unresolved') from None
+
+
+
+class IntakeApprovalReader:
+    """GET-only adapter for present_existing (INFD-WP-0010-T01).
+
+    It has no create, entry or consume route. Any failure is ``None``-equivalent
+    for the caller: intake refuses to present what it could not read.
+    """
+
+    def __init__(self, origin, token_provider, *, transport=None, allow_internal_http=False):
+        self.origin = fixed_origin(origin, allow_internal_http=allow_internal_http)
+        self.token_provider = token_provider
+        self.transport = transport or JSONTransport(allow_internal_http=allow_internal_http)
+
+    def __call__(self, approval_id):
+        if not isinstance(approval_id, str) or not re.fullmatch(
+                r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", approval_id):
+            raise IntakeError(400, "invalid_approval_id")
+        try:
+            token = self.token_provider()
+            if (not isinstance(token, str) or not token or len(token) > 16384
+                    or not token.isascii() or any(c.isspace() for c in token)):
+                raise ValueError()
+            status, result = self.transport.request('GET', self.origin + '/v1/approvals/' + approval_id,
+                headers={'Authorization': 'Bearer ' + token})
+            if status != 200 or not isinstance(result, dict):
+                raise ValueError()
+            return result
+        except Exception:
+            raise IntakeError(503, 'approval_unreadable') from None

informed_decision/intake_service.py

@@ -12,13 +12,18 @@ from pathlib import Path
 import re
 import uuid
 
-from .intake import IntakeRefused, SubmitterGrant, canonical, idempotency_outcome, validate_submission
+from .intake import (IntakeRefused, PresentExisting, SubmitterGrant, canonical, idempotency_outcome,
+                     validate_submission)
 from .memo import Principal
 from .records import dumps
 from .store import Conflict
 
 AUDIENCE = "informed-decision-intake"
 MAX_BODY = 1024 * 1024
+ROW_KEYS = {"submitter_id", "memo_classes", "review_groups", "daily_limit"}
+PRESENT_EXISTING_KEYS = {"approval_actor", "actions"}
+_ACTOR_RE = re.compile(r"service:[a-z0-9][a-z0-9-]{1,62}")
+_ACTION_RE = re.compile(r"[a-z0-9][a-z0-9-]{1,62}")
 
 
 class IntakeError(Exception):
@@ -66,7 +71,8 @@ def configuration(submitters_path, groups_path):
             raise ValueError()
         for username, row in submitters.items():
             if (not re.fullmatch(r"system:serviceaccount:[a-z0-9-]+:[a-z0-9-]+", username)
-                    or set(row) != {"submitter_id", "memo_classes", "review_groups", "daily_limit"}
+                    or not isinstance(row, dict)
+                    or set(row) not in (ROW_KEYS, ROW_KEYS | {"present_existing"})
                     or not isinstance(row["submitter_id"], str)
                     or not row["submitter_id"] or type(row["daily_limit"]) is not int
                     or row["daily_limit"] < 1):
@@ -74,6 +80,15 @@ def configuration(submitters_path, groups_path):
             for field in ("memo_classes", "review_groups"):
                 if not isinstance(row[field], list) or any(not isinstance(x, str) for x in row[field]):
                     raise ValueError()
+            if "present_existing" in row:
+                profile = row["present_existing"]
+                if (not isinstance(profile, dict) or set(profile) != PRESENT_EXISTING_KEYS
+                        or not isinstance(profile["approval_actor"], str)
+                        or not _ACTOR_RE.fullmatch(profile["approval_actor"])
+                        or profile["approval_actor"] == "service:informed-decision"
+                        or not isinstance(profile["actions"], list) or not profile["actions"]
+                        or any(not isinstance(a, str) or not _ACTION_RE.fullmatch(a) for a in profile["actions"])):
+                    raise ValueError()
         if len({r["submitter_id"] for r in submitters.values()}) != len(submitters):
             raise ValueError()
         for row in groups.values():
@@ -84,10 +99,51 @@ def configuration(submitters_path, groups_path):
         raise IntakeError(503, "intake_configuration_unavailable") from None
 
 
+def verify_existing(approval, submission, *, now=None):
+    """present_existing (INFD-WP-0010-T01): may this existing approval be presented?
+
+    Every check must hold; the first failure is the refusal code. Nothing here
+    decides whether the act is permitted: it only refuses to present an object
+    whose binding is not exactly the one the memo names, or that a human entry
+    could no longer complete.
+    """
+    memo = submission.memo
+    expected = submission.approval_binding
+    now = now or datetime.now(timezone.utc)
+    if not isinstance(approval, dict) or approval.get("id") != memo.approval_id:
+        return "approval_mismatch"
+    binding = approval.get("binding")
+    if not isinstance(binding, dict):
+        return "approval_mismatch"
+    if approval.get("status") != "requested":
+        return "approval_not_requested"
+    if approval.get("entries") != []:
+        return "approval_has_entries"
+    if binding.get("human_control") is not True:
+        return "approval_not_human_controlled"
+    if binding.get("actor") != expected["actor"]:
+        return "approval_actor_mismatch"
+    if binding.get("digest") != memo.approval_binding_digest:
+        return "approval_digest_mismatch"
+    if any(binding.get(k) != expected[k] for k in ("action", "principal", "purpose", "target")):
+        return "approval_binding_mismatch"
+    if approval.get("required_count") != 1:
+        return "approval_required_count_unsupported"
+    try:
+        expires = datetime.fromisoformat(approval["validity"]["expires_at"].replace("Z", "+00:00"))
+        if expires.tzinfo is None or expires <= now:
+            return "approval_expired"
+    except (KeyError, TypeError, AttributeError, ValueError):
+        return "approval_expired"
+    return None
+
+
 class IntakeService:
-    def __init__(self, store, authenticator, creator, *, submitters_path, groups_path):
+    def __init__(self, store, authenticator, creator, *, submitters_path, groups_path, reader=None):
+        """``reader(approval_id) -> dict`` is a read-only GET used only by present_existing."""
         self.store, self.authenticator, self.creator = store, authenticator, creator
         self.submitters_path, self.groups_path = submitters_path, groups_path
+        self.reader = reader
 
     def pose(self, token, envelope):
         username = self.authenticator.authenticate(token)
@@ -96,12 +152,21 @@ class IntakeService:
         if row is None:
             raise IntakeError(403, "submitter_unknown")
         grant = SubmitterGrant(row["submitter_id"], frozenset(row["memo_classes"]), frozenset(row["review_groups"]))
-        submission = validate_submission(envelope, submitter_id=grant.submitter_id, grant=grant)
+        profile = None
+        if "present_existing" in row:
+            profile = PresentExisting(row["present_existing"]["approval_actor"],
+                                      frozenset(row["present_existing"]["actions"]))
+        submission = validate_submission(envelope, submitter_id=grant.submitter_id, grant=grant,
+                                         present_existing=profile)
+        if profile is not None and self.reader is None:
+            raise IntakeError(503, "present_existing_unavailable")
         recipient = groups.get(submission.review_group)
         if recipient is None:
             raise IntakeError(503, "review_group_unconfigured")
         metadata = {"memo_class": submission.memo_class, "review_group": submission.review_group,
                     "drafted_by": submission.drafted_by, "serves": submission.serves}
+        if profile is not None:
+            metadata.update(profile="present_existing", approval_requester=profile.approval_actor)
         recovered = None
         # Reserve before any network side effect. SQLite serializes concurrent poses.
         with self.store._transaction() as db:
@@ -115,8 +180,9 @@ class IntakeService:
                     return 200, json.loads(previous["receipt"])
                 if previous["state"] == "created":
                     recovered = json.loads(previous["receipt"])
-                else:
+                elif profile is None:
                     raise IntakeError(409, "approval_create_unresolved")
+                # present_existing has no uncertain side effect: an interrupted read is re-verified.
             if not previous and (db.execute("SELECT 1 FROM memos WHERE id=?", (submission.memo.id,)).fetchone()
                     or db.execute("SELECT 1 FROM intake_ledger WHERE memo_id=?", (submission.memo.id,)).fetchone()):
                 raise IntakeError(409, "memo_id_taken")
@@ -126,12 +192,32 @@ class IntakeService:
                                (submission.submitter_id, day)).fetchone()[0]
             if not previous and count >= row["daily_limit"]:
                 raise IntakeError(429, "submitter_budget_exhausted")
+            if profile is not None and not previous:
+                self._refuse_presented(db, submission.memo.approval_id)
             metadata["day"] = day
             if not previous:
                 db.execute("INSERT INTO intake_ledger VALUES (?,?,?,?,?,?,NULL)",
                            (submission.submitter_id, submission.idempotency_key, submission.submission_digest,
                             submission.memo.id, "creating", dumps(metadata)))
-        if recovered is None:
+        if recovered is None and profile is not None:
+            # Read-only. No create, entry or consume is ever sent in this path.
+            try:
+                approval = self.reader(submission.memo.approval_id)
+            except Exception:
+                approval = None
+            refusal = "approval_unreadable" if approval is None else verify_existing(approval, submission)
+            if refusal is not None:
+                # No engine side effect happened: release the reservation.
+                with self.store._transaction() as db:
+                    db.execute("DELETE FROM intake_ledger WHERE submitter=? AND key=? AND state='creating'",
+                               (submission.submitter_id, submission.idempotency_key))
+                raise IntakeError(503 if refusal == "approval_unreadable" else 409, refusal)
+            approval_id, digest = submission.memo.approval_id, submission.memo.approval_binding_digest
+            receipt = {"memo_id": submission.memo.id, "approval_id": approval_id, "approval_binding_digest": digest}
+            with self.store._transaction() as db:
+                db.execute("UPDATE intake_ledger SET state='created',receipt=? WHERE submitter=? AND key=?",
+                           (dumps(receipt), submission.submitter_id, submission.idempotency_key))
+        elif recovered is None:
             try:
                 approval = self.creator({"binding": submission.approval_binding,
                                          "human_control": True, "required_count": 1})
@@ -162,6 +248,8 @@ class IntakeService:
                                  (submission.submitter_id, submission.idempotency_key)).fetchone()
             if current["state"] == "ready":
                 return 200, json.loads(current["receipt"])
+            if profile is not None:
+                self._refuse_presented(db, approval_id, except_key=(submission.submitter_id, submission.idempotency_key))
             for item in envelope["packet"]:
                 data = submission.packet[item["item_id"]]
                 document_digest = "sha256:" + hashlib.sha256(data).hexdigest()
@@ -187,6 +275,16 @@ class IntakeService:
                        (submission.submitter_id, submission.idempotency_key))
         return 201, receipt
 
+    @staticmethod
+    def _refuse_presented(db, approval_id, except_key=None):
+        """One existing approval is presented by at most one memo."""
+        rows = db.execute("SELECT submitter,key FROM intake_ledger WHERE json_extract(receipt,'$.approval_id')=?",
+                          (approval_id,)).fetchall()
+        if any((r["submitter"], r["key"]) != except_key for r in rows):
+            raise IntakeError(409, "approval_already_presented")
+        if db.execute("SELECT 1 FROM memos WHERE json_extract(body,'$.approval_id')=?", (approval_id,)).fetchone():
+            raise IntakeError(409, "approval_already_presented")
+
     def __call__(self, environ, start_response):
         try:
             if environ.get("PATH_INFO") != "/v1/memos":

informed_decision/requester.py

@@ -20,7 +20,10 @@ class RequesterError(Exception):
 
 
 class RequesterTokenProvider:
-    def __init__(self, issuer, secret_file, *, transport=None, clock=time.time):
+    def __init__(self, issuer, secret_file, *, transport=None, clock=time.time, scope="approval:create"):
+        if scope not in ("approval:create", "approval:read"):
+            raise ValueError("requester scope is create or read only")
+        self.scope = scope
         self.issuer = https_origin(issuer)
         self.secret_file = Path(secret_file)
         if not self.secret_file.is_absolute():
@@ -44,7 +47,7 @@ class RequesterTokenProvider:
                 headers={"Content-Type": "application/x-www-form-urlencoded",
                          "Authorization": "Basic " + credentials},
                 body=urlencode({"grant_type": "client_credentials",
-                                "scope": "approval:create"}).encode())
+                                "scope": self.scope}).encode())
             if (status != 200 or not isinstance(result.get("token_type"), str)
                     or result["token_type"].lower() != "bearer"
                     or "refresh_token" in result):
@@ -71,7 +74,7 @@ class RequesterTokenProvider:
                     or claims.get("tenant") != "tenant:platform"
                     or claims.get("principal_type") != "service"
                     or claims.get("roles") != [CLIENT_ID]
-                    or claims.get("scope") != "approval:create"
+                    or claims.get("scope") != self.scope
                     or type(claims["iat"]) is not int or type(claims["exp"]) is not int
                     or claims["iat"] > now or claims["exp"] <= now
                     or not 0 < claims["exp"] - claims["iat"] <= 900):
@@ -88,9 +91,13 @@ def compose_intake(store, token_review_call, *, issuer, secret_file,
     This factory grants no submitter and starts no socket. The separate private
     listener and deployment remain reviewed owner configuration.
     """
-    from .intake_approval import IntakeApprovalClient
+    from .intake_approval import IntakeApprovalClient, IntakeApprovalReader
     from .intake_service import IntakeService, TokenReview
     provider = RequesterTokenProvider(issuer, secret_file, transport=transport)
     creator = IntakeApprovalClient(approval_origin, provider, allow_internal_http=True)
+    # present_existing only; needs approval:read on the requester client (KeyCape registration).
+    # Until granted, that exchange is refused and only present_existing submitters fail closed.
+    read_provider = RequesterTokenProvider(issuer, secret_file, transport=transport, scope="approval:read")
+    reader = IntakeApprovalReader(approval_origin, read_provider, allow_internal_http=True)
     return IntakeService(store, TokenReview(token_review_call), creator,
-                         submitters_path=submitters_path, groups_path=groups_path)
+                         submitters_path=submitters_path, groups_path=groups_path, reader=reader)

informed_decision/store.py

@@ -282,7 +282,9 @@ class Store:
                 or receipt['approval_binding_digest'] != memo.approval_binding_digest):
             raise Conflict('intake receipt does not match memo')
         return {'review_group': metadata['review_group'], 'memo_class': metadata['memo_class'],
-                'intake_submitter': row['submitter'], 'approval_requester': 'informed-decision',
+                'intake_submitter': row['submitter'],
+                # present_existing names the approval's own actor; the create path is informed-decision.
+                'approval_requester': metadata.get('approval_requester', 'informed-decision'),
                 'approval_human_control': True}
 
     def intake_context(self, memo):