From asi
Transforms Gmail messages into GF(3)-typed interactions, routes to triadic queues, detects saturation for inbox-zero state. For email triage, workflow automation, ANIMA application.
npx claudepluginhub plurigrid/asi --plugin asiThis skill uses the workspace's default tool permissions.
Transform Gmail into an ANIMA-condensed system with GF(3) conservation.
Unifies Google Workspace management (Gmail, Drive, Calendar, Tasks, Docs) via ACSet schema and GF(3)-typed interactions. For multi-service automation workflows.
Executes Gmail triage actions: archive, delete, label, star, unsubscribe inbox emails; drafts replies from vault reports; backfills calendar events. Use /gmail for cleanup or drafting.
Scans unread emails from Gmail or Hey.com, scores by priority (VIP, urgency, deadlines), classifies, saves relevant as vault notes, generates triage report.
Share bugs, ideas, or general feedback.
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
Gmail ANIMA applies the ANIMA framework to email:
┌────────────────────────────────────────────────────────────────────┐
│ 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 │
└────────────────────────────────────────────────────────────────────┘
| Object | Description | Trit Role |
|---|---|---|
Interaction | Single email action with verb + trit | Data |
Thread | Gmail conversation with saturation state | Aggregate |
Agent3 | Queue fiber (MINUS/ERGODIC/PLUS) | Router |
QueueItem | Links Interaction → Agent3 | Edge |
Person | Email contact | Node |
Partner | Relationship edge in contact graph | Edge |
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,
}
| Tool | Trit | Description |
|---|---|---|
search_gmail_messages | -1 | Search inbox (MINUS) |
get_gmail_message_content | -1 | Read message (MINUS) |
get_gmail_thread_content | -1 | Read thread (MINUS) |
list_gmail_labels | -1 | List labels (MINUS) |
modify_gmail_message_labels | 0 | Change labels (ERGODIC) |
batch_modify_gmail_message_labels | 0 | Bulk labels (ERGODIC) |
send_gmail_message | +1 | Send email (PLUS) |
draft_gmail_message | +1 | Create draft (PLUS) |
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, ...] │
│ │
└─────────────────────────────────────────┘
agent_of(i) = route(trit(i))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 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
)
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.
Proofs in src/narya_proofs/:
def prove_queue_consistency(system: TriadicQueueSystem) -> bool:
"""Verify no duplication and route invariant."""
return (
system.verify_no_duplication() and
system.verify_route_invariant()
)
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
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
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
| File | Description | Trit |
|---|---|---|
| gmail_acset.py | ACSet schema + GF(3) thread tracking | 0 |
| anima_detector.py | Saturation + equilibrium detection | 0 |
| gmail_mcp_bridge.py | MCP tool wiring with guards | 0 |
| triadic_queues.py | Three disjoint queue fibers | 0 |
| narya_proofs/ | Formal verification proofs | -1 |
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.")
# 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(...)
# 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']}")
# 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']}")
# 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
| Skill | Trit | Integration |
|---|---|---|
| google-workspace | 0 | MCP tool provider |
| gay-mcp | +1 | SplitMixTernary RNG |
| sheaf-cohomology | -1 | H¹ obstruction verification |
| bisimulation-game | -1 | State equivalence proofs |
| ordered-locale | 0 | Thread ordering topology |
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 ✓
Gmail-ANIMA integrates with the full workspace via WorkspaceACSet:
| Morphism | Target | Trigger | GF(3) Effect |
|---|---|---|---|
thread_file | DriveFile | Attachment detected | 0 (ERGODIC) |
thread_event | CalendarEvent | Meeting scheduled | +1 (PLUS) |
thread_task | Task | Action item identified | +1 (PLUS) |
# 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
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
| File | Description |
|---|---|
| workspace_acset.py | Unified schema with cross-skill morphisms |
| mcp_api_equivalence.py | MCP↔API behavioral equivalence |
| path_invariance.py | Workflow path verification |
| workflow_validator.py | End-to-end validation |
| ANIMA Concept | Gmail Implementation |
|---|---|
| Saturation | Thread trit sum ≡ 0 (mod 3) |
| Condensation | Equivalence class collapse |
| MaxEnt Default | needs_action=False initially |
| Persistence | Only flip when forced |
| Replay Invariance | Schedule-independent fingerprint |
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
This skill connects to the K-Dense-AI/claude-scientific-skills ecosystem:
general: 734 citations in bib.duckdbThis skill maps to Cat# = Comod(P) as a bicomodule in the equipment structure:
Trit: 0 (ERGODIC)
Home: Prof
Poly Op: ⊗
Kan Role: Adj
Color: #26D826
The skill participates in triads satisfying:
(-1) + (0) + (+1) ≡ 0 (mod 3)
This ensures compositional coherence in the Cat# equipment structure.