ISO/IEC 27001:2022 legacy integration patterns for enterprise AI agents — Reference Architecture

Reference architecture · published 2026-10-16 · regulated enterprise · ISO/IEC 27001:2022 · layout walkthrough · implementation executed in CI before publish

The control that resolves this is an Idempotent System-of-Record Bridge, evaluated before the request reaches a model provider rather than audited afterwards. In regulated enterprise the binding constraint is ISO/IEC 27001:2022, and the question it actually asks is is this risk on the register with a named owner and a treatment. The specific problem here is mainframe integration. The shape below holds p95 ≤ 2400ms over scheduled batch file exchange (SFTP), and ships with a standard-library implementation that runs as written.

Why this blocks production

Agents can read the legacy estate and cannot safely write to it. A retry after a gateway timeout creates a duplicate posting, and in an ERP a duplicate journal entry is a finance incident. So the team that owns the system of record withholds write access, and the automation stalls at read-only.

What makes mainframe integration specifically hard

The mainframe is the system of record and its change window is measured in quarters. Anything that requires a change on that side is not an integration plan, it is a request to join a release queue.

What resolves it is a side-car integration that requires zero mainframe change. The failure to design against: a design whose first step is a CICS change nobody will schedule.

What ISO/IEC 27001:2022 specifically requires

ISO 27001 is a management-system standard: the auditor checks that risks are identified, owned, treated and reviewed on a cycle. A perfect technical control that is absent from the risk register is a nonconformity.

The artifact that satisfies it is risk register entries with owners, treatments and review dates. The way teams most often fail it: a technically sound control that no Statement of Applicability references, so it cannot be evidenced.

Component breakdown

1. Content-derived idempotency key — Derived from business keys, so two attempts at one intent collide.
2. Claim table — Transactional claim in the same database as the outbox; a retry observes the first attempt.
3. Typed write contract — Allow-listed fields per document type; unknown fields are rejected loudly.
4. Outbox — Durable intent record, so a crash mid-flight is recoverable rather than ambiguous.
5. Reconciliation job — Compares posted documents against intents and surfaces drift.
6. Compensating action — A defined reversal path per document type, because some writes must be undoable.

Reference implementation — Idempotent write-back with a typed write contract

Standard-library Python, no external dependencies, runs as written. Executed by the publish gate: if it raises, this page does not ship. Wire it behind scheduled batch file exchange (SFTP); the failure mode it guards against is a partial file processed as complete. The output you hand the assessor is the Statement of Applicability extract.

"""Idempotent write-back adapter for a legacy system of record (SAP/ERP shape).

Why this shape: the blocker on legacy integration is almost never reading. It
is writing. An agent that retries a posting because a gateway timed out will
happily create a duplicate journal entry, and in an ERP a duplicate posting is
a finance incident, not a bug report.

Three controls make agent-driven writes acceptable to the team that owns the
system of record:
  * a deterministic idempotency key derived from business content
  * a claim table, so a retry observes the first attempt instead of repeating it
  * a bounded, typed projection, so the agent can never post a field it does
    not understand
"""
import hashlib
import json
import time
from dataclasses import dataclass


class DuplicatePosting(Exception):
    """Raised when a retry would create a second document for one intent."""


def idempotency_key(doc_type: str, business_keys: dict) -> str:
    """Derived from BUSINESS content, never from a uuid4.

    A random key makes every retry unique, which is the opposite of what
    idempotency means. Two attempts to post the same invoice must collide.
    """
    canonical = json.dumps({k: business_keys[k] for k in sorted(business_keys)},
                           sort_keys=True, separators=(",", ":"))
    return f"{doc_type}:" + hashlib.sha256(canonical.encode()).hexdigest()[:32]


ALLOWED_FIELDS = {
    "journal_entry": {"company_code", "posting_date", "currency",
                      "debit_account", "credit_account", "amount", "reference"},
}


def project(doc_type: str, payload: dict) -> dict:
    """Allow-list projection. An agent-proposed field that is not in the
    contract is dropped loudly, not posted hopefully."""
    allowed = ALLOWED_FIELDS.get(doc_type)
    if allowed is None:
        raise ValueError(f"no write contract defined for {doc_type!r}")
    unknown = set(payload) - allowed
    if unknown:
        raise ValueError(f"fields outside the write contract: {sorted(unknown)}")
    missing = allowed - set(payload)
    if missing:
        raise ValueError(f"write contract requires: {sorted(missing)}")
    return {k: payload[k] for k in sorted(allowed)}


@dataclass
class ClaimStore:
    """Stands in for a real transactional table. In production this is a row
    in the SAME database as your outbox, claimed with SELECT ... FOR UPDATE."""

    def __post_init__(self):
        self._claims: dict = {}

    def claim(self, key: str, ttl: float = 900.0) -> tuple:
        """-> (is_new, existing_result). Atomic in a real DB."""
        now = time.time()
        rec = self._claims.get(key)
        if rec and (rec["status"] == "done" or now - rec["at"] < ttl):
            return False, rec
        self._claims[key] = {"status": "in_flight", "at": now, "result": None}
        return True, None

    def complete(self, key: str, result: dict):
        self._claims[key] = {"status": "done", "at": time.time(), "result": result}

    def fail(self, key: str):
        self._claims.pop(key, None)      # let a genuine retry through


def post_document(store: ClaimStore, doc_type: str, business_keys: dict,
                  payload: dict, transport) -> dict:
    """The only sanctioned path from an agent to the system of record."""
    key = idempotency_key(doc_type, business_keys)
    body = project(doc_type, payload)

    is_new, existing = store.claim(key)
    if not is_new:
        if existing["status"] == "done":
            # The correct response to a duplicate is the ORIGINAL result.
            return {"idempotency_key": key, "replayed": True,
                    **existing["result"]}
        raise DuplicatePosting(f"{key} is already in flight")

    try:
        result = transport(doc_type, body, key)
    except Exception:
        store.fail(key)
        raise
    store.complete(key, result)
    return {"idempotency_key": key, "replayed": False, **result}


if __name__ == "__main__":
    calls = []

    def fake_sap(doc_type, body, key):
        calls.append(key)
        return {"document_number": f"50000{len(calls)}", "posted": True}

    store = ClaimStore()
    keys = {"invoice_id": "INV-8831", "company_code": "1000"}
    payload = {"company_code": "1000", "posting_date": "2026-10-16",
               "currency": "EUR", "debit_account": "400000",
               "credit_account": "113100", "amount": "14500.00",
               "reference": "INV-8831"}

    first = post_document(store, "journal_entry", keys, payload, fake_sap)
    retry = post_document(store, "journal_entry", keys, payload, fake_sap)
    print("first :", first["document_number"], "replayed", first["replayed"])
    print("retry :", retry["document_number"], "replayed", retry["replayed"])
    print("transport invocations:", len(calls), "(must be 1)")
    try:
        project("journal_entry", {**payload, "tax_code": "V1"})
    except ValueError as e:
        print("contract rejects unknown field:", e)

The ISO/IEC 27001:2022 control, as code

Statement of Applicability with operating evidence. This is the artifact an assessor asks for under ISO/IEC 27001:2022 specifically — it is not the same control as the one above, and a system that implements one does not thereby implement the other. It executes under the same publish gate.

"""Annex A control to evidence mapping behind a Statement of Applicability.

ISO/IEC 27001:2022 turns on the SoA: every Annex A control is either applicable
with a justification and evidence, or excluded with a justification. The audit
failure is an applicable control whose evidence is a policy document rather
than a record that the control operated.
"""

ANNEX_A_2022 = {
    "A.5.23": "Information security for use of cloud services",
    "A.8.2": "Privileged access rights",
    "A.8.10": "Information deletion",
    "A.8.12": "Data leakage prevention",
    "A.8.15": "Logging",
    "A.8.16": "Monitoring activities",
    "A.8.24": "Use of cryptography",
    "A.8.28": "Secure coding",
}

EVIDENCE_KINDS = {"record", "configuration", "test_result"}
NON_EVIDENCE = {"policy", "intent", "roadmap"}


class StatementOfApplicability:
    def __init__(self, scope: str):
        self.scope = scope
        self.entries: dict[str, dict] = {}

    def apply(self, control: str, justification: str,
              evidence_ref: str, evidence_kind: str) -> None:
        if control not in ANNEX_A_2022:
            raise KeyError(f"{control} is not an Annex A:2022 control")
        if evidence_kind in NON_EVIDENCE:
            raise ValueError(
                f"{control}: {evidence_kind!r} describes intent, not operation")
        if evidence_kind not in EVIDENCE_KINDS:
            raise ValueError(f"{control}: unknown evidence kind {evidence_kind!r}")
        self.entries[control] = {"applicable": True, "justification": justification,
                                 "evidence_ref": evidence_ref,
                                 "evidence_kind": evidence_kind}

    def exclude(self, control: str, justification: str) -> None:
        if control not in ANNEX_A_2022:
            raise KeyError(f"{control} is not an Annex A:2022 control")
        if len(justification.split()) < 5:
            raise ValueError(f"{control}: exclusion needs a real justification")
        self.entries[control] = {"applicable": False, "justification": justification}

    def unaddressed(self) -> list[str]:
        return sorted(set(ANNEX_A_2022) - set(self.entries))

    def certifiable(self) -> dict:
        gaps = self.unaddressed()
        applied = [c for c, e in self.entries.items() if e["applicable"]]
        return {"scope": self.scope, "applied": sorted(applied),
                "excluded": sorted(c for c, e in self.entries.items()
                                   if not e["applicable"]),
                "unaddressed": gaps, "certifiable": not gaps}


if __name__ == "__main__":
    soa = StatementOfApplicability("LLM platform, production tenant")
    soa.apply("A.5.23", "Model inference is a cloud service",
              "cfg://gateway/allowed-providers.yaml", "configuration")
    soa.apply("A.8.2", "Operators hold privileged access to prompts",
              "rec://iam/quarterly-access-review-2026Q3", "record")
    soa.apply("A.8.10", "Retention is enforced on prompt logs",
              "test://retention/expiry-suite-2026-10", "test_result")
    soa.apply("A.8.12", "Egress to model providers can leak data",
              "cfg://dlp/egress-classifier.yaml", "configuration")
    soa.apply("A.8.15", "All inference events are logged",
              "rec://audit/chain-head-2026-10-16", "record")
    soa.apply("A.8.16", "Anomalous usage is monitored",
              "rec://siem/llm-usage-alerts", "record")
    soa.apply("A.8.24", "TLS and at-rest encryption",
              "test://crypto/tls-config-scan", "test_result")
    soa.exclude("A.8.28", "No application code is developed within this scope "
                          "boundary; the platform is configured, not coded")

    report = soa.certifiable()
    assert report["certifiable"], report
    assert len(report["applied"]) == 7 and len(report["excluded"]) == 1, report

    try:
        soa.apply("A.8.15", "we have a logging policy",
                  "doc://policies/logging.pdf", "policy")
        raise AssertionError("accepted a policy document as operating evidence")
    except ValueError:
        pass

    partial = StatementOfApplicability("partial")
    partial.apply("A.8.15", "logging", "rec://x", "record")
    assert not partial.certifiable()["certifiable"]
    assert len(partial.certifiable()["unaddressed"]) == 7

    print("ISO/IEC 27001:2022 SoA evidence mapping: OK")
    print(f"  applied {len(report['applied'])}, excluded {len(report['excluded'])}, "
          f"unaddressed {len(report['unaddressed'])}")

System architecture

Idempotent System-of-Record Bridge — the request path, with the decision points that must exist before the first model call:

flowchart LR
    AG[Agent] --> PR[Projection<br/>allow-listed fields]
    PR -->|unknown field| REJ[Reject - loud]
    PR --> IK[Idempotency key<br/>sha256 of business keys]
    IK --> CL{Claim table}
    CL -->|already done| RP[Replay original result]
    CL -->|in flight| DUP[Refuse duplicate]
    CL -->|new| OB[(Outbox)]
    OB --> TR[Transport - OData / IDoc / RFC]
    TR --> SOR[(System of record<br/>SAP / ERP / AS400)]
    SOR --> RC[Reconciliation]
    RC -->|drift| CMP[Compensating action]

Technical parameters

Every value below is a parameter you can hold a vendor or an internal team to. Adjectives are not acceptance criteria.

Risk register with acceptance criteria

Each row carries an acceptance parameter rather than an adjective, because "low latency" cannot be signed off. Retention here is bound by per the data class retention schedule, enforced automatically.

What the auditor will ask

• Where is this control in the risk register and the Statement of Applicability, who owns it, and when was it last reviewed?
  Register entries are generated from the deployed control set, so every control has an SoA reference, a named owner and a review date that cannot silently lapse.
• What does this require the mainframe team to change, and when is their next window?
  Nothing. All new behaviour lives in the side-car and the existing interface is consumed as-is, so delivery does not depend on their release queue.
• If the gateway times out and the agent retries, how many documents exist — and what proves it?
  One. The idempotency key is derived from the business keys, so both attempts collide on the same claim; the retry observes the completed claim and replays the original result. The transport is invoked once, which the test asserts.
• Which exact fields may an agent write per document type, and what happens to a field outside that list?
  Only the fields in that document type's allow-list, and all of them — the projection rejects both unknown fields and incomplete payloads. A field outside the contract raises rather than posting hopefully.
• What does the system-of-record owner see when a posting is wrong, and what reverses it?
  The reconciliation job compares posted documents against recorded intents and surfaces drift; each writable document type has a defined, rehearsed compensating action for reversal.
• Where is the intent recorded before the transport call, and what resolves a mid-flight crash?
  In the outbox, written before the transport call. On restart, reconciliation resolves anything left in flight, so a crash leaves a recoverable state rather than an ambiguous one.
• Is the extract pulling the minimum field set, with a justification per field?
  It should be — projection applies on read as well as write, with a recorded justification per field, so the extract schema matches the documented minimum field set.

Related architectures

• Part of legacy system integration with ai agents.
• HIPAA legacy integration patterns for enterprise AI agents applies the same mechanism under a different obligation.
• For the related constraint, see SOC 2 Type II legacy integration patterns for enterprise AI agents.
• This pattern also governs EU AI Act legacy integration patterns for enterprise AI agents.
• PCI DSS 4.0 legacy integration patterns for enterprise AI agents applies the same mechanism under a different obligation.
• If this is the blocker in front of you: enterprise AI infrastructure review.