Segmented Persistence and Forensic Traceability

VOXAgentMemory vs VOXAgentStore

VOX explicitly separates two persistence concerns at the agent level:

Aspect	VOXAgentMemory	VOXAgentStore
Purpose	Immutable forensic ledger	Operational asset storage
File	agents/<name>/memory/logs.db	agents/<name>/memory/memory.db
Data type	Events, actions, audit trail	Files, metadata, custom tables
Mutability	Append-only (INSERT only)	Read/write with updates
Structure	Fixed activity_log table	asset_index table + dynamic tables via create_table()
Indexes	timestamp, event_type, ref_id	asset_type, checksum, name

Design Rationale

The separation serves three criteria:

1. Forensic performance: the append-only ledger can scale to millions of records without fragmentation, with timestamp indexes for time-window queries. The operational store requires random writes and updates that would fragment the ledger if they shared the same file.

2. Security: the ledger is immutable by design — once inserted, a record cannot be modified or deleted. The operational store allows deletions and updates (e.g., delete_file, last_accessed). Keeping them separate prevents a store compromise from corrupting the forensic chain.

3. Lifecycle: the store can be purged or reindexed without affecting the audit trail. The ledger is permanent.

---

Append-Only Forensic Ledger

activity_log Table Schema

CREATE TABLE IF NOT EXISTS activity_log (
    id TEXT PRIMARY KEY,            -- UUID v4
    timestamp REAL NOT NULL,        -- time.time() (Unix epoch, microseconds)
    event_type TEXT NOT NULL,       -- Event category
    actor TEXT NOT NULL,            -- "SYSTEM" | "<agent_name>" | ...
    action TEXT NOT NULL,           -- Action description
    details TEXT NOT NULL DEFAULT '{}',  -- JSON event payload
    ref_id TEXT,                    -- Correlation UUID (e.g., message_id)
    status TEXT NOT NULL DEFAULT 'COMPLETED'  -- COMPLETED | PENDING | RUNNING | FAILED
);

CREATE INDEX idx_timestamp ON activity_log(timestamp);
CREATE INDEX idx_event_type ON activity_log(event_type);
CREATE INDEX idx_ref_id ON activity_log(ref_id);

Public Methods

Method	Description
record(event_type, action, actor, details, ref_id, status)	Inserts an immutable record, returns the generated UUID
get_recent(limit=100)	Last N records ordered by descending timestamp
get_thread(ref_id)	All records in a correlation chain (ascending)
get_pending()	Records with status PENDING or RUNNING (for recovery)

Forensic Trace Example

ID: a1b2c3d4-...
Timestamp: 1747350000.123456
EventType: "command_execution"
Actor: "root"
Action: "delegated_command"
Details: {"command": "analyze_cve", "target": "<coordinator_uuid>", "cve_id": "CVE-2024-..."}
RefID: "op-msg-001"
Status: "COMPLETED"

ID: e5f6g7h8-...
Timestamp: 1747350001.234567
EventType: "command_execution"
Actor: "coordinator"
Action: "delegate_to_subordinate"
Details: {"command": "analyze_cve", "target": "<specialist_uuid>", "original_source": "root"}
RefID: "op-msg-001"
Status: "COMPLETED"

ID: i9j0k1l2-...
Timestamp: 1747350100.345678
EventType: "task_result"
Actor: "specialist"
Action: "task_completed"
Details: {"result": "...", "severity": "HIGH", "cve_id": "CVE-2024-..."}
RefID: "op-msg-001"
Status: "COMPLETED"

DFIR reconstruction: calling get_thread("op-msg-001") returns the complete causality chain — from the original command emission (root) to the result (specialist), including intermediate delegation (coordinator).

Implementation

async def record(self, event_type, action, actor="SYSTEM", details=None,
                 ref_id=None, status="COMPLETED") -> str:
    event_id = str(uuid.uuid4())
    try:
        async with aiosqlite.connect(self._db_path) as conn:
            await conn.execute(
                "INSERT INTO activity_log (id, timestamp, event_type, actor, action, details, ref_id, status) "
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                (event_id, time.time(), event_type, actor, action,
                 json.dumps(details or {}), ref_id, status),
            )
            await conn.commit()
    except Exception as e:
        logger.exception("Forensic record failed [%s]: %s", event_type, e)
    return event_id

The method catches exceptions but always returns the generated event_id even if the INSERT fails, to avoid interrupting the agent flow.

---

VOXAgentStore — Operational Storage

asset_index Table Schema

CREATE TABLE IF NOT EXISTS asset_index (
    id TEXT PRIMARY KEY,            -- UUID v4
    archived_at REAL NOT NULL,      -- Unix archive timestamp
    name TEXT NOT NULL,             -- Original filename
    asset_type TEXT NOT NULL,       -- Type: "screenshot", "document", "evidence", ...
    origin TEXT NOT NULL,           -- Origin system
    file_path TEXT NOT NULL UNIQUE, -- Relative path in assets/
    file_size INT NOT NULL,         -- Size in bytes
    checksum TEXT NOT NULL,         -- SHA-256 of content
    last_accessed REAL,             -- Last access (for garbage collection)
    tags TEXT NOT NULL DEFAULT '[]' -- JSON array of tags
);

File Storage

The asset sandbox lives at agents/<name>/assets/. Each file is stored with a UUID name to prevent collisions and directory traversal:

def store_file(self, data, filename, origin, asset_type, tags=None):
    checksum = hashlib.sha256(data).hexdigest()
    # Deduplication by checksum
    existing = self.query("SELECT * FROM asset_index WHERE checksum = ?", (checksum,))
    if existing and "error" not in existing[0]:
        return {**existing[0], "duplicate": True}
    # Physical storage with UUID name
    stored_name = f"{asset_id}{Path(filename).suffix}"
    path = self._assets_dir / stored_name
    path.write_bytes(data)

Search and Deduplication

• Search by asset_type, name (LIKE), tags (JSON contains)
• Automatic deduplication by SHA-256: if the same file already exists, returns the existing record with a duplicate: true flag
• Access control: get_safe_path() prevents directory traversal

Sandbox Security

def get_safe_path(self, sub_dir, filename):
    sandbox_root = (self.dir / "assets").resolve()
    target_dir = (sandbox_root / sub_dir).resolve()
    clean_name = filename.replace("/", "")
    final_path = (target_dir / clean_name).resolve()
    if not str(final_path).startswith(str(sandbox_root)):
        raise PermissionError("Sandbox escape attempt")
    return final_path

---

Forensic Value for DFIR

The combination of VOXAgentMemory (ledger) + VOXAgentStore (assets) + VOXMessage (traceable envelope) provides:

1. Chain of custody: every message, command, and decision is recorded with a UUID and atomic timestamp in the ledger of each involved agent
2. Incident reconstruction: get_thread(ref_id) reconstructs the complete event sequence of an incident
3. Multi-agent correlation: ref_id propagates through the hierarchy, enabling reconstruction of the root→coordinator→specialist flow
4. Preserved evidence: screenshots, documents, and outputs are stored with SHA-256 checksums in the asset sandbox, immutable
5. Dual logging: VOXForensicLogger writes simultaneously to colored console (operators) and flat file (forensic) with VOXLogSource metadata