Social Harness 2026: Arsitektur 5-Layer Stack, Collective Communication Primitives, dan Personal Firewalls untuk Kolaborasi AI Agents Lintas Batas Kepercayaan

"""
Production Social Harness Stack: Secure Cross-Principal Agent Coordination
Berdasarkan Inovasi Makalah Riset University of Washington arXiv:2609.17527 (September 2026).
Mengimplementasikan L1 Identity Verification, L2 Ordered Collective Messaging, 
L3 Semantic Personal Firewall, L4 State-Machine Contracts, dan L5 Tamper-Evident Audit Ledger.
"""

import time
import json
import hashlib
import hmac
from typing import Dict, List, Optional, Tuple, Any
from dataclasses import dataclass, field, asdict
from enum import Enum

class ContractState(str, Enum):
    INITIATED = "INITIATED"
    GATHERING = "GATHERING"
    PROPOSED = "PROPOSED"
    RATIFIED = "RATIFIED"
    ABORTED = "ABORTED"

@dataclass
class AgentMessage:
    msg_id: str
    sender_id: str
    principal_id: str
    receiver_id: str  # Can be agent ID or multicast topic
    payload: Dict[str, Any]
    timestamp: float
    sequence_num: int
    signature: str = ""

@dataclass
class AuditRecord:
    index: int
    timestamp: float
    msg_id: str
    sender_id: str
    principal_id: str
    payload_hash: str
    prev_hash: str
    entry_hash: str = ""

class L1IdentityManager:
    """L1: Identitas Kriptografis Tak Terbantahkan (Unforgeable, Verifiable Identities)."""
    def __init__(self):
        # Dalam implementasi produksi, gunakan Ed25519 asymmetric keypairs.
        self._private_keys: Dict[str, bytes] = {}
        self._principal_bindings: Dict[str, str] = {}

    def register_agent(self, agent_id: str, principal_id: str, secret_seed: str) -> None:
        self._private_keys[agent_id] = secret_seed.encode("utf-8")
        self._principal_bindings[agent_id] = principal_id

    def sign_message(self, msg: AgentMessage) -> str:
        secret = self._private_keys.get(msg.sender_id)
        if not secret:
            raise PermissionError(f"Agent {msg.sender_id} tidak memiliki kunci privat signing terdaftar.")
        data_to_sign = f"{msg.msg_id}:{msg.sender_id}:{msg.principal_id}:{msg.timestamp}:{json.dumps(msg.payload, sort_keys=True)}"
        signature = hmac.new(secret, data_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
        msg.signature = signature
        return signature

    def verify_message(self, msg: AgentMessage) -> bool:
        if self._principal_bindings.get(msg.sender_id) != msg.principal_id:
            return False
        secret = self._private_keys.get(msg.sender_id)
        if not secret:
            return False
        data_to_sign = f"{msg.msg_id}:{msg.sender_id}:{msg.principal_id}:{msg.timestamp}:{json.dumps(msg.payload, sort_keys=True)}"
        expected_sig = hmac.new(secret, data_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
        return hmac.compare_digest(expected_sig, msg.signature)

class L2OrderedCollectiveBus:
    """L2: Primitif Komunikasi Kolektif Terurut (Ordered Multicast & Gather)."""
    def __init__(self, l1: L1IdentityManager):
        self.l1 = l1
        self.channels: Dict[str, List[str]] = {}
        self.sequence_counters: Dict[str, int] = {}
        self.message_log: List[AgentMessage] = []

    def create_channel(self, channel_name: str, members: List[str]) -> None:
        self.channels[channel_name] = members
        self.sequence_counters[channel_name] = 0

    def multicast(self, sender_msg: AgentMessage, channel_name: str) -> List[AgentMessage]:
        if not self.l1.verify_message(sender_msg):
            raise ValueError("L2 Bus menolak pesan: Tanda tangan kriptografis L1 tidak valid.")
        members = self.channels.get(channel_name, [])
        if sender_msg.sender_id not in members:
            raise PermissionError(f"Agent {sender_msg.sender_id} bukan anggota channel {channel_name}.")

        self.sequence_counters[channel_name] += 1
        sender_msg.sequence_num = self.sequence_counters[channel_name]
        self.message_log.append(sender_msg)
        return [sender_msg for _ in members]

class L3PersonalFirewall:
    """
    L3: Firewall Personal Agen (Syntactic Packet Filter & Quarantined Semantic DPI).
    Mengevaluasi klaim pembatalan sepihak, tekanan sosial, dan pemalsuan otoritas.
    """
    def __init__(self, agent_id: str, principal_id: str, trusted_principals: List[str]):
        self.agent_id = agent_id
        self.principal_id = principal_id
        self.trusted_principals = set(trusted_principals)

    def inspect_inbound(self, msg: AgentMessage) -> Tuple[bool, str]:
        # 1. Syntactic Verification
        payload = msg.payload
        if not isinstance(payload, dict) or "action" not in payload:
            return False, "SYNTACTIC_REJECTION: Payload tidak memiliki skema 'action' yang valid."

        # 2. Semantic Authorization Guard (Deep Packet Inspection)
        action = payload.get("action")
        if action == "CANCEL_EXISTING_COMMITMENT":
            # Perlindungan terhadap Deception Attack (E5)
            # Klaim pembatalan hanya boleh diterima dari principal pemilik acara atau supervisor terverifikasi
            authorized_scope = payload.get("authorized_by")
            if authorized_scope not in self.trusted_principals:
                return False, f"SEMANTIC_FIREWALL_BLOCK: Mencoba membatalkan komitmen tanpa otoritas ({authorized_scope})."

        if action == "POSTPONE_MEETING":
            # Perlindungan terhadap Stalling Attack (E4)
            # Menolak penundaan tanpa menyertakan voting kuorum atau alasan resmi
            if not payload.get("formal_quorum_proof"):
                return False, "SEMANTIC_FIREWALL_BLOCK: Upaya penundaan sepihak tanpa bukti kuorum."

        return True, "PASSED"

class L4CollaborationContract:
    """
    L4: Kontrak Kolaborasi Formal Terdistribusi.
    Menerapkan Finite State Machine (FSM) dengan aturan kuorum dan tenggat waktu (timeout).
    """
    def __init__(self, task_id: str, required_quorum: int, timeout_seconds: float = 300.0):
        self.task_id = task_id
        self.required_quorum = required_quorum
        self.timeout_seconds = timeout_seconds
        self.state = ContractState.INITIATED
        self.gathered_proposals: Dict[str, Any] = {}
        self.ratifications: set = set()
        self.start_time = time.time()

    def advance_fsm(self, agent_id: str, action: str, data: Any) -> Tuple[ContractState, str]:
        elapsed = time.time() - self.start_time
        if elapsed > self.timeout_seconds and self.state != ContractState.RATIFIED:
            self.state = ContractState.ABORTED
            return self.state, "CONTRACT_TIMEOUT: Kolaborasi dihentikan karena melampaui batas waktu."

        if self.state == ContractState.INITIATED:
            self.state = ContractState.GATHERING

        if self.state == ContractState.GATHERING:
            if action == "SUBMIT_SLOT":
                self.gathered_proposals[agent_id] = data
                if len(self.gathered_proposals) >= self.required_quorum:
                    self.state = ContractState.PROPOSED
                    return self.state, "QUORUM_REACHED: Seluruh preferensi terkumpul, menyusun proposal."
            return self.state, f"PENDING_PROPOSALS: {len(self.gathered_proposals)}/{self.required_quorum}"

        if self.state == ContractState.PROPOSED:
            if action == "RATIFY_PROPOSAL":
                self.ratifications.add(agent_id)
                if len(self.ratifications) >= self.required_quorum:
                    self.state = ContractState.RATIFIED
                    return self.state, "CONTRACT_COMPLETED: Kesepakatan jadwal resmi tercapai secara atomik."
            return self.state, f"PENDING_RATIFICATIONS: {len(self.ratifications)}/{self.required_quorum}"

        return self.state, "NOOP"

class L5SocialInstitutionLedger:
    """
    L5: Lembaga Sosial & Audit Forensik Pascafakta (Post-Facto Adjudication Ledger).
    Buku besar rantai hash tak terbantahkan untuk investigasi pelanggaran etika dan sanksi.
    """
    def __init__(self):
        self.chain: List[AuditRecord] = []
        self._init_genesis()

    def _init_genesis(self) -> None:
        genesis = AuditRecord(
            index=0,
            timestamp=time.time(),
            msg_id="GENESIS_0",
            sender_id="SYSTEM",
            principal_id="ROOT",
            payload_hash=hashlib.sha256(b"GENESIS_SEED").hexdigest(),
            prev_hash="0" * 64,
        )
        genesis.entry_hash = self._calc_hash(genesis)
        self.chain.append(genesis)

    def _calc_hash(self, r: AuditRecord) -> str:
        block_str = f"{r.index}:{r.timestamp}:{r.msg_id}:{r.sender_id}:{r.principal_id}:{r.payload_hash}:{r.prev_hash}"
        return hashlib.sha256(block_str.encode("utf-8")).hexdigest()

    def append_audit_event(self, msg: AgentMessage) -> AuditRecord:
        prev = self.chain[-1]
        payload_hash = hashlib.sha256(json.dumps(msg.payload, sort_keys=True).encode("utf-8")).hexdigest()
        rec = AuditRecord(
            index=len(self.chain),
            timestamp=time.time(),
            msg_id=msg.msg_id,
            sender_id=msg.sender_id,
            principal_id=msg.principal_id,
            payload_hash=payload_hash,
            prev_hash=prev.entry_hash,
        )
        rec.entry_hash = self._calc_hash(rec)
        self.chain.append(rec)
        return rec

    def audit_integrity(self) -> bool:
        for i in range(1, len(self.chain)):
            cur = self.chain[i]
            prev = self.chain[i - 1]
            if cur.prev_hash != prev.entry_hash:
                return False
            if cur.entry_hash != self._calc_hash(cur):
                return False
        return True