AI Agent Logging for the CRA’s 24-Hour Early Warning

By /
Screenprint illustration of a hand posting a sealed envelope into a clock-faced filing box on a desk, with a European civic square behind

9 min read · 2,256 words

This article was written with AI. It was drafted from the sources it cites and checked against the full text of those sources before publishing. How we make articles

The Cyber Resilience Act (CRA) now gives manufacturers 24 hours to file an early warning with ENISA once they are aware a vulnerability is being actively exploited, and Dark Reading reports that failing to report serious cybersecurity incidents could cost up to €15 million or 2.5% of worldwide annual revenue, whichever is higher (Dark Reading). This tutorial builds that record: an AI agent logging pipeline, from schema to evidence export.

What You’ll Build

By the end, you will have a working AI agent logging pipeline built entirely on the Python standard library: a SQLite audit database, a middleware wrapper that records every tool call with UTC timestamps, actor identity, and model version, an awareness-event table that separates the first signal from legal awareness, a script that computes your 24- and 72-hour deadlines and states the final-report rule, an evidence exporter producing JSONL plus SHA-256 manifests, and a dry-run drill that exercises the whole chain.

Timing matters here. Reporting duties took effect 11 September 2026, while conformity assessments for new products wait until 11 December 2027 (JD Supra). An estimated 25,000 companies fall in scope (Yahoo News), and one survey on the eve of the deadline found 66% of organisations aware of the CRA at all (Legal IT Insider). Cross those figures and, if the survey rate holds for in-scope companies, roughly 8,500 of them (25,000 × 0.34, the share the survey found unaware) started the clock blind. This instrumentation is yours to build.

Prerequisites

  • Python 3.11 or newer (everything below uses only the standard library; no pip installs)
  • sqlite3 CLI 3.35+ (optional, for inspecting the database)
  • An agent loop where tools are ordinary Python callables
  • A Linux or macOS terminal, about 25 minutes
  • One piece of context worth loading first: the Commission’s implementation guidance C(2026) 5252 does not mention AI agents a single time, according to coverage of the document (Forkast). What the guidance does not define, you have to instrument.

Step 1: Initialize the Agent Log Database

Every defense of a late filing eventually reduces to what was recorded and when. Start with a schema that answers both questions.

Create audit_schema.py:

import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS agent_events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    ts_utc TEXT NOT NULL,
    actor TEXT NOT NULL,
    model_id TEXT,
    session_id TEXT NOT NULL,
    tool_name TEXT NOT NULL,
    target TEXT,
    args_digest TEXT,
    outcome TEXT NOT NULL,
    error_detail TEXT
);
CREATE INDEX IF NOT EXISTS idx_events_session ON agent_events(session_id);
CREATE INDEX IF NOT EXISTS idx_events_ts ON agent_events(ts_utc);

CREATE TABLE IF NOT EXISTS awareness_events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    opened_utc TEXT NOT NULL,
    certain_utc TEXT,
    product TEXT NOT NULL,
    summary TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'open'
);
"""

def init(path="agent_audit.db"):
    conn = sqlite3.connect(path)
    conn.executescript(SCHEMA)
    conn.commit()
    return conn

if __name__ == "__main__":
    init()
    print("schema ready")

Run it:

python audit_schema.py
sqlite3 agent_audit.db ".tables"

Success looks like the script printing schema ready, a new agent_audit.db file in the working directory, and .tables listing agent_events and awareness_events. The two-table split is deliberate: raw tool calls are high-volume and low-meaning until triage turns one of them into an awareness event. Keeping them separate keeps the evidentiary record small and readable.

Step 2: Wrap Tool Calls With the Audit Middleware

Create agent_wrapper.py:

import hashlib
import json
import sqlite3
import uuid
from datetime import datetime, timezone

from audit_schema import init

_conn = init()

def _now():
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds")

def audited_tool_call(session_id, actor, model_id, tool_name, target, args, fn):
    digest = hashlib.sha256(
        json.dumps(args, sort_keys=True, default=str).encode()
    ).hexdigest()
    row = (_now(), actor, model_id, session_id, tool_name, target,
           digest, None, None)
    try:
        result = fn(**args)
    except Exception as exc:
        row = row[:-2] + ("error", f"{type(exc).__name__}: {exc}")
        _write(row)
        raise
    row = row[:-2] + ("ok", None)
    _write(row)
    return result

def _write(row):
    _conn.execute(
        "INSERT INTO agent_events (ts_utc, actor, model_id, session_id,"
        " tool_name, target, args_digest, outcome, error_detail)"
        " VALUES (?,?,?,?,?,?,?,?,?)", row)
    _conn.commit()

if __name__ == "__main__":
    import os, tempfile

    def write_note(text):
        path = os.path.join(tempfile.gettempdir(), "audit_demo.txt")
        with open(path, "w") as fh:
            fh.write(text)
        return path

    audited_tool_call(
        session_id=str(uuid.uuid4()),
        actor="svc-agent-runner",
        model_id="demo-agent-1.0",
        tool_name="write_note",
        target="workspace://audit_demo.txt",
        args={"text": "hello audit trail"},
        fn=write_note,
    )
    print("event recorded")

$ python drill.py
[… 4 lines omitted …]
  final report due: 14 days after a corrective measure is available (vulnerabilities), or one month (severe incidents)
event #2 product=acme-gateway-2.4 status=confirmed
  anchor (certain or first signal): 2026-09-26T04:26:05.700000+00:00
  early warning due: 2026-09-27T04:26:05.700000+00:00
  notification due: 2026-09-29T04:26:05.700000+00:00
  final report due: 14 days after a corrective measure is available (vulnerabilities), or one month (severe incidents)
{'file': 'evidence/6c36037d-0aa9-478a-81fd-3b3287fa94c1.jsonl', 'sha256': 'f93fc8aeae850407323cee2854acc557723590a10dadf4097ee4d31f2e290bea', 'events': 1}
ls: cannot access '/tmp/drill_target.txt': No such file or directory
6c36037d-0aa9-478a-81fd-3b3287fa94c1.jsonl
6c36037d-0aa9-478a-81fd-3b3287fa94c1.manifest.json
83524d7f-f02b-4045-ace1-4bede615b3cc.jsonl
83524d7f-f02b-4045-ace1-4bede615b3cc.manifest.json
Output of our own test run of the code above, recorded 2026-09-26.

Run python agent_wrapper.py, then verify:

sqlite3 agent_audit.db "SELECT ts_utc, actor, model_id, tool_name, outcome FROM agent_events;"

A successful run prints event recorded, and the query returns one row with the actor and model values you passed in. Note the design trade-off: arguments are stored as a SHA-256 digest, not raw content. You lose replayability and gain data minimization, which matters once a security log starts containing customer data. Keep raw arguments, if at all, in a separate store with its own retention policy and access review. Each agent_events row still records tool_name, target and ts_utc next to args_digest: the record you need once the 24-hour early-warning clock starts (European Commission).

Step 3: Record the Awareness Event With Two Timestamps

Under Commission guidance, as JD Supra summarizes it, a manufacturer becomes aware “once, following an initial assessment, it has a reasonable degree of certainty that a vulnerability in its product is being actively exploited or that a severe incident has compromised the security of its product” (JD Supra). First alert and legal awareness are different moments, and your schema must capture both. Call the design a two-timestamp awareness model: first signal in one column, reasonable certainty in the other, and every deadline downstream anchored to the second one whenever it exists.

Create awareness.py:

import sqlite3
from datetime import datetime, timezone

def _now():
    return datetime.now(timezone.utc).isoformat(timespec="milliseconds")

def open_event(product, summary, db="agent_audit.db"):
    conn = sqlite3.connect(db)
    cur = conn.execute(
        "INSERT INTO awareness_events (opened_utc, product, summary, status)"
        " VALUES (?,?,?,?)",
        (_now(), product, summary, "open"))
    conn.commit()
    return cur.lastrowid

def mark_certain(event_id, db="agent_audit.db"):
    conn = sqlite3.connect(db)
    conn.execute(
        "UPDATE awareness_events SET certain_utc=?, status='confirmed'"
        " WHERE id=?", (_now(), event_id))
    conn.commit()

Why two timestamps: the gap between opened_utc and certain_utc is your triage window, and defending that window is the whole game. On this article’s reading, a certain_utc set after the first log line documents the triage window; without one, the only recorded anchor for the deadline is the first signal.

Verify by opening the Python REPL, calling open_event("demo-product", "triage test"), then mark_certain(1), and confirming both columns hold distinct UTC timestamps with sqlite3 agent_audit.db "SELECT * FROM awareness_events;".

Step 4: Generate the 24/72-Hour Deadline Report

The statutory windows are fixed: early warning within 24 hours of awareness, full notification within 72 hours, and a final report no later than 14 days after a corrective measure becomes available for actively exploited vulnerabilities, or within one month from the 72-hour notification for severe incidents (European Commission). Compute them, do not eyeball them. Both clock windows share one anchor, which leaves 48 hours between the early warning deadline and the notification deadline (72 − 24). That gap, not the full 72, is the working window your team actually gets once the first filing is out.

Create deadlines.py:

import sqlite3
from datetime import datetime, timedelta

def deadline_report(db="agent_audit.db"):
    conn = sqlite3.connect(db)
    conn.row_factory = sqlite3.Row
    rows = conn.execute(
        "SELECT id, product, status, opened_utc, certain_utc"
        " FROM awareness_events WHERE status != 'closed' ORDER BY id"
    ).fetchall()
    if not rows:
        print("no open awareness events")
        return
    for r in rows:
        anchor = datetime.fromisoformat(r["certain_utc"] or r["opened_utc"])
        print(f"event #{r['id']} product={r['product']} status={r['status']}")
        print(f"  anchor (certain or first signal): {anchor.isoformat()}")
        print(f"  early warning due: {(anchor + timedelta(hours=24)).isoformat()}")
        print(f"  notification due: {(anchor + timedelta(hours=72)).isoformat()}")
        # the final report runs from the fix, not from awareness: record it when a fix ships
        print("  final report due: 14 days after a corrective measure is available"
              " (vulnerabilities), or one month (severe incidents)")

Run python -c "from deadlines import deadline_report; deadline_report()". Success prints one block per open event with its computed early-warning and notification due times, both ending in +00:00, and the final-report rule, which runs from the fix. According to the European Commission, filings go through ENISA’s Single Reporting Platform, submitted once, routed to the CSIRT of the main establishment and, unless particularly exceptional circumstances apply, made available simultaneously to ENISA (European Commission). Your script’s job ends at producing the dates and the evidence; a human still files.

Step 5: Export Portal-Ready Evidence Bundles

Forkast reported that the platform launches without an API, so teams fill out its forms manually, in English (Forkast). So automate the part worth automating: evidence assembly.

Create export_evidence.py:

import hashlib
import json
import os
import sqlite3

def export(session_id, db="agent_audit.db", out_dir="evidence"):
    os.makedirs(out_dir, exist_ok=True)
    conn = sqlite3.connect(db)
    conn.row_factory = sqlite3.Row
    rows = conn.execute(
        "SELECT * FROM agent_events WHERE session_id=? ORDER BY ts_utc",
        (session_id,)).fetchall()
    path = os.path.join(out_dir, f"{session_id}.jsonl")
    with open(path, "w") as fh:
        for r in rows:
            fh.write(json.dumps(dict(r), sort_keys=True) + "\n")
    with open(path, "rb") as fh:
        digest = hashlib.sha256(fh.read()).hexdigest()
    manifest = {"file": path, "sha256": digest, "events": len(rows)}
    mpath = os.path.join(out_dir, f"{session_id}.manifest.json")
    with open(mpath, "w") as fh:
        json.dump(manifest, fh, indent=2)
    return manifest

Call export("<session-id>") and check the evidence/ directory. A successful export leaves a .jsonl file and a .manifest.json carrying its hash and event count. The manifest is what turns a log dump into evidence: anyone can recompute the hash and confirm the bundle has not been edited since export.

Step 6: Run the Dry-Run Drill

Compliance tooling that has never executed under pressure is a hypothesis. This drill simulates the scenario that keeps platform teams awake: an agent deleting a file outside its sandbox.

Create drill.py:

import os
import tempfile
import uuid

from audit_schema import init
from agent_wrapper import audited_tool_call
from awareness import open_event, mark_certain
from deadlines import deadline_report
from export_evidence import export

init()
session = str(uuid.uuid4())
target = os.path.join(tempfile.gettempdir(), "drill_target.txt")
with open(target, "w") as fh:
    fh.write("placeholder")

def delete_file(path):
    os.remove(path)

audited_tool_call(
    session_id=session,
    actor="svc-agent-runner",
    model_id="drill-model-0.9",
    tool_name="delete_file",
    target=target,
    args={"path": target},
    fn=delete_file,
)

event_id = open_event(
    "acme-gateway-2.4",
    "agent deleted a file outside its sandbox; assessing active exploitation")
mark_certain(event_id)
deadline_report()
print(export(session))

Run python drill.py in the same directory. Success looks like: a deadline report that now includes the drill’s event (event #2, after the Step 3 test) with early-warning and notification due times 24 and 72 hours after its anchor, a drill_target.txt that no longer exists, and one more JSONL file plus its manifest in evidence/. Every printed timestamp should end in +00:00.

One scope note before moving on: these reporting duties cover products with digital elements already placed on the market before 11 December 2027, not just new shipments (JD Supra). The guidance also confirms there is no obligation to report exploitation a manufacturer already knew about before September 11, 2026 (JD Supra).

Common Pitfalls

Local timestamps. The code forces timezone.utc on every write for a reason. Because the script computes each deadline from the stored timestamp, a timestamp written in local time instead of UTC shifts every deadline by the local offset. Never patch in datetime.now() without timezone.utc.

Logging raw prompts by default. The digest-first design exists so a security log does not silently become a customer-data store. If an investigation needs raw arguments, pull them from a separate, access-controlled store with short retention. Running a privacy review over the log schema costs an afternoon; running it during an incident costs the incident.

SQLite lock contention. Multiple agent workers writing concurrently can hit SQLITE_BUSY (database is locked), which SQLite returns when a write is blocked by concurrent activity from another connection (SQLite). Fix it with two settings:

PRAGMA journal_mode=WAL;
PRAGMA busy_timeout=5000;

journal_mode=WAL is persistent, so setting it once is enough (SQLite). Write-ahead logging lets readers proceed while a writer commits.

Trying to automate the filing. The portal has no API at launch (Forkast). Build the pipeline to the export step and stop. Do not build a cron job that files to an endpoint that does not exist.

Misreading the exemptions. Dark Reading reports that microenterprises and small enterprises may not be fined for missing the 24-hour window (Dark Reading). Treat that as relief for the finance team, not a reason to skip the log. Open-source software is out of CRA scope (Dark Reading), though open-source maintainers may still want the same log for their own triage.

What’s Next

Ship the JSONL bundles to your SIEM and alert when outcome='error' events spike: a run of failed tool calls can be an early sign that an agent is doing something it should not. For the December 2027 conformity side, on Sept. 10, 2026, the Eclipse Foundation announced a free toolkit developed through the EU-funded OCCTET project, validated across ten technology ecosystems, with a PurlDB demonstrator covering more than 20 million packages (Business Insider Markets), which pairs well with this audit layer. The Commission’s own guidance ships 67 practical examples and flowcharts worth mapping against your schema (European Commission). When guidance and blog posts disagree, go back to the statute itself, Regulation (EU) 2024/2847 (EUR-Lex).

References

Leave a Comment

Scroll to Top