Gmail Anima

Gmail ANIMA Skill

Transform Gmail into an ANIMA-condensed system with GF(3) conservation.

Trit: 0 (ERGODIC - coordinator)
Principle: Inbox Zero = Condensed Equilibrium State
Implementation: GmailACSet + TriadicQueues + AnimaDetector

Overview

Gmail ANIMA applies the ANIMA framework to email:

1. Transform - Messages → GF(3)-typed Interactions
2. Route - Interactions → Triadic queue fibers (MINUS/ERGODIC/PLUS)
3. Detect - Saturation → ANIMA condensed state
4. Verify - Narya proofs for consistency

GmailACSet Schema

┌────────────────────────────────────────────────────────────────────┐
│                      GmailACSet Schema                             │
├────────────────────────────────────────────────────────────────────┤
│                                                                    │
│  Interaction ─────┬────▶ Thread                                   │
│  ├─ verb: String  │      ├─ thread_id: String                     │
│  ├─ timebin: Int  │      ├─ needs_action: Bool                    │
│  ├─ trit: Trit    │      ├─ last_action_bin: Int                  │
│  └─ person ───────┼──▶   └─ saturated: Bool                       │
│                   │                                                │
│  QueueItem ───────┼────▶ Agent3                                   │
│  ├─ interaction ──┘      ├─ fiber: Trit {-1, 0, +1}               │
│  └─ agent ───────────▶   └─ name: String                          │
│                                                                    │
│  Person ◀─────────────── Partner ────────────────▶ Person         │
│  ├─ email: String        ├─ src                                   │
│  └─ name: String         ├─ tgt                                   │
│                          └─ weight: Int                            │
└────────────────────────────────────────────────────────────────────┘

Objects

GF(3) Verb Typing

Gmail actions are assigned trits based on information flow:

VERB_TRIT_MAP = {
    # MINUS (-1): Consumption/Validation
    "read": -1,      "search": -1,     "view": -1,
    "fetch": -1,     "list": -1,
    
    # ERGODIC (0): Coordination/Metadata
    "label": 0,      "archive": 0,     "snooze": 0,
    "star": 0,       "mark_read": 0,   "mark_unread": 0,
    "move": 0,
    
    # PLUS (+1): Generation/Execution
    "send": +1,      "forward": +1,    "reply": +1,
    "schedule": +1,  "draft": +1,      "compose": +1,
}

MCP Tool → Trit Mapping

Triadic Queue Routing

Interactions route to disjoint queue fibers:

                    ┌─────────────────────────────────────────┐
                    │           TRIADIC QUEUES                │
                    ├─────────────────────────────────────────┤
                    │                                         │
   Interaction ────▶│  route(trit) ───▶ Agent3 Fiber         │
                    │                                         │
                    │  MINUS (-1)  ────▶ [read, search, ...]  │
                    │  ERGODIC (0) ────▶ [label, archive, ...]│
                    │  PLUS (+1)   ────▶ [send, reply, ...]   │
                    │                                         │
                    └─────────────────────────────────────────┘

Invariants

1. No duplication: Each interaction in exactly one fiber
2. Route invariant: agent_of(i) = route(trit(i))
3. Ordering: PLUS must be preceded by MINUS in same thread
4. Conservation: Thread trit sum ≡ 0 (mod 3) at cycle close

Queue Depth Balance

def saturation_metrics(queues: Dict[Agent3, deque]) -> Dict:
    depths = [len(q) for q in queues.values()]
    return {
        'balance_ratio': min(depths) / max(depths),  # 1.0 = perfect
        'gf3_residue': sum(i.trit for q in queues for i in q) % 3,
    }

Saturation Detection → ANIMA State

Saturation occurs when a thread reaches stable equilibrium:

def is_saturated(thread_id: str) -> bool:
    """Thread is saturated when:
    1. No change in needs_action for N steps
    2. GF(3) cycle closure: sum(trits) ≡ 0 (mod 3)
    3. History window shows identical states
    """
    history = detector.history[thread_id][-N:]
    cycle_sum = sum(t for t in thread.gf3_cycle[-3:])
    
    return (
        all(s == history[0] for s in history) and  # Stable
        (cycle_sum % 3) == 0                        # Conserved
    )

ANIMA Detection

def detect_anima() -> Dict:
    """System at ANIMA when:
    1. All threads saturated
    2. GF(3) conserved globally
    3. Equivalence classes stable
    4. Replay invariance holds
    """
    return {
        "at_anima": all_saturated and gf3_conserved and stable_impacts,
        "condensed_fingerprint": sha256(sorted_equiv_classes),
        "persistence_bars_stable": True,
    }

Inbox Zero as ANIMA: When all threads reach saturation with GF(3) conservation, the inbox is in condensed equilibrium.

Narya Proof Integration

Proofs in src/narya_proofs/:

1. Queue Consistency (queue_consistency.py)

def prove_queue_consistency(system: TriadicQueueSystem) -> bool:
    """Verify no duplication and route invariant."""
    return (
        system.verify_no_duplication() and
        system.verify_route_invariant()
    )

2. Replay Determinism (replay_determinism.py)

def prove_replay_determinism(schedule1, schedule2) -> bool:
    """Different schedules → identical condensed state."""
    fp1 = replay(schedule1).condensed_fingerprint
    fp2 = replay(schedule2).condensed_fingerprint
    return fp1 == fp2

3. Non-Leakage (non_leakage.py)

def prove_non_leakage(bridge: GmailMCPBridge) -> bool:
    """No interaction leaks between fibers."""
    for agent, queue in bridge.queues.items():
        for item in queue:
            if bridge._route(item.trit) != agent:
                return False
    return True

4. GF(3) Conservation (gf3_conservation.py)

def prove_gf3_conservation(bridge: GmailMCPBridge) -> bool:
    """All closed cycles satisfy sum ≡ 0 (mod 3)."""
    for cycle in bridge.cycle_tracker.closed_cycles:
        if sum(cycle.trits) % 3 != 0:
            return False
    return True

Source Files

Workflows

Workflow 1: Triage Inbox to ANIMA

from gmail_mcp_bridge import create_gmail_bridge
from anima_detector import AnimaDetector

# Create bridge
bridge = create_gmail_bridge("user@gmail.com")
detector = AnimaDetector(saturation_threshold=5)

# MINUS: Read unread messages
bridge.search_gmail_messages("is:unread")
for msg in results:
    bridge.get_gmail_message_content(msg.id, thread_id=msg.thread_id)
    detector.update_thread(msg.thread_id, trit=Trit.MINUS)

# ERGODIC: Label/archive processed
for msg in processed:
    bridge.modify_gmail_message_labels(
        msg.id,
        add_label_ids=["Label_Processed"],
        remove_label_ids=["INBOX"],
        thread_id=msg.thread_id
    )
    detector.update_thread(msg.thread_id, trit=Trit.ERGODIC)

# Check ANIMA
anima = detector.detect_anima()
if anima["at_anima"]:
    say("Inbox at ANIMA. Condensed state achieved.")

Workflow 2: Reply with GF(3) Guard

# MINUS first: Read the thread
bridge.get_gmail_thread_content(thread_id)  # trit=-1

# PLUS: Reply (requires prior MINUS)
try:
    bridge.send_gmail_message(
        to="reply@example.com",
        subject="Re: Topic",
        body="Response...",
        thread_id=thread_id,
        in_reply_to=original_message_id
    )  # trit=+1
except GF3ConservationError:
    # Must read before sending
    bridge.get_gmail_thread_content(thread_id)  # Retry after MINUS
    bridge.send_gmail_message(...)

Workflow 3: Batch Triage with Saturation

# Create balanced batch
batch = create_triadic_batch(
    payloads=["read_1", "label_1", "archive_1"],  # Will balance to 0
    thread_id="batch_thread",
    seed=1069
)

system = TriadicQueueSystem()
for interaction in batch:
    if system.enqueue(interaction):
        print(f"✓ {interaction.payload} → {interaction.agent.name}")

# Check metrics
stats = system.full_statistics()
print(f"GF(3) Residue: {stats['saturation']['gf3_residue']}")  # 0
print(f"Cycles Closed: {stats['operations']['cycles_closed']}")

Workflow 4: Sheaf Cohomology Verification

# After processing
h1 = bridge.verify_h1_obstruction()
print(f"H¹ obstructions: {h1['h1']}")
print(f"Globally consistent: {h1['globally_consistent']}")

# Obstructions = threads not at GF(3) = 0
for v in h1['violations']:
    print(f"  Thread {v['thread_id']}: residue={v['mod_3']}")

Commands

# Run Gmail ANIMA demo
python src/gmail_acset.py

# Test triadic queues
python src/triadic_queues.py

# Run ANIMA detector
python src/anima_detector.py

# Run Narya proofs
python -m src.narya_proofs.runner

Integration with Other Skills

GF(3) Triadic Conservation

gmail-anima (0) ⊗ sheaf-cohomology (-1) ⊗ gay-mcp (+1) = 0 ✓
gmail-anima (0) ⊗ bisimulation-game (-1) ⊗ send (+1) = 0 ✓
read (-1) ⊗ label (0) ⊗ reply (+1) = 0 ✓

Cross-Skill Integration

Gmail-ANIMA integrates with the full workspace via WorkspaceACSet:

Morphisms from Gmail

Workflow Paths

# Gmail → Task (balanced)
path = gmail_read >> task_create  # -1 + 1 = 0 ✓

# Full workflow (needs balancing)
full = gmail_read >> drive_create >> calendar_create >> task_create
balanced = balance_path(full)  # Auto-adds ERGODIC steps

MCP ↔ API Equivalence

Gmail operations can be executed via MCP tools or direct API:

# Equivalent executions
mcp_result = bridge.execute_mcp("send_gmail_message", params)
api_result = bridge.execute_api("gmail_send", params)
assert mcp_result.state == api_result.state

Source Files (Extended)

ANIMA Principles Applied

Say Narration Integration

from gmail_mcp_bridge import NaryaLogger

logger = NaryaLogger(voice="Ava (Premium)")

# Announces: "Gmail bridge: MINUS transition"
logger.log(before, after, Trit.MINUS, impact=False)

# Announces: "Gmail bridge: PLUS transition, impact detected"
logger.log(before, after, Trit.PLUS, impact=True)

Skill Name: gmail-anima
Type: Email Management / ANIMA Framework
Trit: 0 (ERGODIC - coordinator)
GF(3): Conserved via triadic queue routing
ANIMA: Inbox Zero = Condensed Equilibrium State

Converted and distributed by TomeVault — claim your Tome and manage your conversions.