Supercharging Local AI Agents on Signal: Fast-Paths, Live Typing Indicators, and Session Continuity

Running a full-scale agentic AI—like Google Antigravity (agy), Claude Code, or a local LLM with tool-use capabilities—directly inside your homelab or developer workstation gives you immense power. You can orchestrate deployments, query infrastructure metrics, check smart home sensors, and run automated diagnostics without exposing any private data or API keys to third-party consumer chat wrappers.

Connecting such an agent to an encrypted mobile messenger like Signal seems like the logical next step: an ultra-secure, pocket-sized command center right in your pocket.

Yet out of the box, interacting with an autonomous AI agent over mobile messaging can be an agonizing experience.

Recently, while away from my desk, I sent a quick message to my personal Signal assistant:

“Why is this interface so slow? Any way to make it more responsive? Looks like there is a big delay in communication.”

The message was met with deafening silence. Thirty-five seconds passed with zero indication on my phone screen whether the server had crashed, signal-cli had dropped the connection, or the agent was even thinking. Then, suddenly, a wall of text popped up.

In mobile communication, perceived latency is user experience. A 30-second delay without a typing indicator feels like a complete system outage. Moreover, forcing an LLM with full tool-calling overhead to cold-boot just to answer “What’s the weather in the garden?” or “Is the UPS battery charged?” is a massive waste of compute and time.

To fix this once and for all, I completely redesigned and hardened my homelab’s Signal gateway (signal_bot.py). In this post, I will walk you through the architecture, latency profiling, and implementation of:

1. Dual-Path Routing: Sub-second (<1s) local deterministic fast-paths vs. background LLM agent inference.
2. Real-Time Asynchronous Typing Indicators: Reverse-engineering signal-cli JSON-RPC to provide instant (<100ms) visual feedback.
3. Sliding Session Continuity: Retaining conversational working memory across mobile turns without bloated prompt re-parsing.
4. Latency Engineering: Tuning agent execution flags (--effort low, --mode auto, strict 5-minute timeout bounds) to drop cold LLM turns from 45s+ to 4–8s.
5. Production Hardening: Single-user “Note to Self” filtering and systemd user daemon deployment.

1. The Bottlenecks of Headless Mobile Agents

Why do local AI agents feel sluggish over messaging platforms? Profiling the standard pipeline revealed four critical bottlenecks:

1. The Cold-Start Tax: When an agent CLI like agy starts, it initializes credential caches, discovers dynamic workspace tools, loads system rules from ~/.agents/AGENTS.md, and runs initial environment checks. Even on modern NVMe drives and fast CPUs, cold startup consumes 3 to 6 seconds before a single token is generated.
2. Deep Reasoning Overhead: For open-ended coding tasks, deep chain-of-thought thinking is essential. But for simple chat inquiries, high reasoning effort causes the model to spend 20+ seconds contemplating nuances that the user never asked for.
3. The “Black Hole” UX: Unlike desktop web interfaces that stream server-sent events (SSE) token-by-token, a standard messaging gateway typically waits for the subshell command to finish before sending the final string payload back over the wire. During that 30–45s window, the mobile UI displays nothing.
4. Episodic Amnesia: Without explicit session chaining, every message is treated as a fresh conversation. If you text “What was the temperature again?” after asking about the weather, the agent has no idea what you are referring to unless you re-feed the entire history.

2. Dual-Path Architecture: Determinism Meets Agency

The first architectural decision was separating deterministic queries from agentic reasoning.

If a user texts “ping”, “status”, “ups”, or “vreme” (weather in Romanian), we do not need a trillion-parameter neural network to write a shell command, parse the output, and draft an essay. We already have battle-tested local CLI scripts in ~/scripts/ that run in less than 50 milliseconds.

Path A: The Instant Fast-Path (<1s)

When a message arrives over the signal-cli JSON-RPC stream, the Python daemon first runs a clean regex parser against an in-memory lookup table of fast-path commands. If matched:

• The command executes directly in a local lightweight Python or Bash process (subprocess.run).
• The output is formatted and sent back to Signal via sendReceipt and a plain JSON-RPC reply.
• Total round-trip time: 300ms to 850ms. The response arrives on your phone almost before your thumb leaves the keyboard.

Path B: Deep Agent Inference (4–8s)

If the prompt does not match any deterministic pattern (e.g., “Help me debug the failing cron on nas2”, or “Summarize the recent changes to the blog post”), the daemon routes the request to Google Antigravity (agy).

• A background TypingPulser thread starts pulsing typing notifications to Signal immediately (<100ms).
• The agent runs with session continuity and optimized effort parameters.
• When execution finishes, the typing pulser halts, and the final response is delivered.

3. Real-Time Typing Indicators over JSON-RPC

In messaging apps like Signal, WhatsApp, and Telegram, visual typing indicators (the animated three bouncing dots) provide crucial psychological feedback: I heard you, I’m working on it, stay tuned.

How signal-cli JSON-RPC Handles Typing

signal-cli has a high-performance daemon mode that accepts JSON-RPC 2.0 requests over standard input (stdin) and writes incoming messages to standard output (stdout).

While many developers use simple one-off CLI calls like signal-cli send, running it in persistent JSON-RPC mode allows you to invoke the sendTyping RPC method without process restart overhead:

{
  "jsonrpc": "2.0",
  "method": "sendTyping",
  "params": {
    "recipient": ["+15551234567"],
    "stop": false
  },
  "id": "typing-1"
}

Signal clients automatically expire a typing indicator after roughly 10 to 15 seconds if no new pulse is received. Therefore, if an agent query takes 25 seconds, a single typing notification at the start will vanish halfway through, returning the user to the “black hole” silence.

The Background TypingPulser Thread

To solve this, we implemented a dedicated, daemonized TypingPulser thread in Python. When an AI turn begins, the pulser fires immediately, and then continues pulsing every 8 seconds until the agent finishes:

import threading
import time
import json
import logging

class TypingPulser:
    """Sends periodic sendTyping pulses over signal-cli JSON-RPC during inference."""
    def __init__(self, rpc_proc, recipient, interval=8.0):
        self.rpc_proc = rpc_proc
        self.recipient = recipient
        self.interval = interval
        self._stop_event = threading.Event()
        self._thread = None

    def start(self):
        self._stop_event.clear()
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def _send_typing(self, stop=False):
        if not self.rpc_proc or self.rpc_proc.poll() is not None:
            return
        msg = {
            "jsonrpc": "2.0",
            "method": "sendTyping",
            "params": {
                "recipient": [self.recipient],
                "stop": stop
            },
            "id": f"type-{int(time.time() * 1000)}"
        }
        try:
            payload = json.dumps(msg) + "\n"
            self.rpc_proc.stdin.write(payload.encode("utf-8"))
            self.rpc_proc.stdin.flush()
        except Exception as e:
            logging.debug("Failed to send typing pulse: %s", e)

    def _run(self):
        # Pulse immediately on trigger (<100ms)
        self._send_typing(stop=False)
        while not self._stop_event.wait(self.interval):
            self._send_typing(stop=False)

    def stop(self):
        self._stop_event.set()
        if self._thread and self._thread.is_alive():
            self._thread.join(timeout=1.0)
        # Send explicit stop signal to clear the typing dots
        self._send_typing(stop=True)

The moment you hit “Send” in Signal, your phone screen reflects active typing in under 100 milliseconds.

4. Sliding Session Continuity

When using an agent CLI from the command line, sessions are tied to your terminal. But messaging is inherently episodic: you ask a question, read a response, lock your phone, and reply two minutes later.

Without session tracking, an agent must re-evaluate your environment and workspace on every turn. In Antigravity (agy), previous sessions can be resumed using the --conversation <CONVERSATION_ID> flag.

We engineered a sliding-window session state manager:

SESSION_STATE_FILE = Path.home() / ".gemini/antigravity-cli/signal_conversation.json"
SESSION_TTL_SECONDS = 7200  # 2-hour sliding window

def get_active_conversation_id() -> str | None:
    if not SESSION_STATE_FILE.exists():
        return None
    try:
        data = json.loads(SESSION_STATE_FILE.read_text())
        updated_at = data.get("updated_at", 0)
        if time.time() - updated_at < SESSION_TTL_SECONDS:
            return data.get("conversation_id")
    except Exception:
        pass
    return None

def set_active_conversation_id(cid: str):
    payload = {
        "conversation_id": cid,
        "updated_at": time.time()
    }
    SESSION_STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    SESSION_STATE_FILE.write_text(json.dumps(payload))

How It Operates

1. If an active conversation exists and was updated within the last 2 hours, signal_bot.py invokes:agy --conversation <CID> --effort low --mode auto "User prompt"
2. If the session has expired, or if the user sends an explicit /new or /reset command, the cached file is cleared, and a fresh session begins.
3. If agy ever returns an error indicating that the conversation ID is stale or corrupted, the handler automatically clears the cache and seamlessly retries as a fresh session without failing the user’s message.

This gives the user continuous, natural dialogue flow right from their phone without manual session management.

5. Latency Profiling: Before vs. After

To quantify the performance improvements, we benchmarked typical user interactions across three categories:

1. Deterministic Fast-Paths (ping, weather, ups).
2. Warm Multi-Turn Reasoning (follow-up query within active session).
3. Cold LLM Startup (fresh conversation requiring full agent bootstrap).

The Benchmarking Numbers

By configuring --effort low and --mode auto, the agent skips exhaustive chain-of-thought pondering for straightforward questions, focusing directly on synthesis and tool execution.

6. Implementation Deep Dive: Fast-Path Handlers

Here is how the fast-path dispatcher is structured in signal_bot.py:

SCRIPTS_DIR = Path.home() / "scripts"

def handle_fast_command(text: str) -> str | None:
    """Interprets deterministic operational commands without LLM cold-start."""
    cmd = text.strip().lower()
    
    if cmd in ("ping", "/ping"):
        return f"🏓 Pong! Host {socket.gethostname()} is active and responsive."
        
    if cmd in ("vreme", "weather", "/weather"):
        res = subprocess.run(
            ["bash", str(SCRIPTS_DIR / "get_ha_sensor.sh"), "sensor.outdoor_temperature"],
            capture_output=True, text=True, timeout=5
        )
        temp = res.stdout.strip() or "N/A"
        return f"🌤️ Outdoor Temperature: {temp}°C"
        
    if cmd in ("ups", "/ups"):
        res = subprocess.run(
            ["bash", str(SCRIPTS_DIR / "get_ups_status.sh")],
            capture_output=True, text=True, timeout=8
        )
        return f"⚡ UPS Power Status:\n{res.stdout.strip()}"
        
    if cmd in ("unifi", "/unifi"):
        res = subprocess.run(
            ["bash", str(SCRIPTS_DIR / "get_unifi_status.sh")],
            capture_output=True, text=True, timeout=10
        )
        return f"🌐 Network Gateway Health:\n{res.stdout.strip()}"

    if cmd in ("new", "/new", "reset", "/reset"):
        clear_active_conversation()
        return "🔄 Conversation session reset. Starting fresh context."

    return None

If handle_fast_command(text) returns a non-null string, the bot sends the reply immediately and returns. Only unmatched messages spin up the TypingPulser and call agy.

7. Production Hardening: Security & Systemd

When hosting an autonomous AI agent connected to root-capable scripts and local infrastructure, security is paramount.

Single-User “Note to Self” Filtering

The bot runs against my personal Signal account linked via signal-cli. Because signal-cli receives messages for all contacts and groups associated with the account, we enforce strict sender verification:

import os

# Filter strictly to your own phone number or account UUID
AUTHORIZED_NUMBER = os.environ.get("AUTHORIZED_PHONE_NUMBER", "+15551234567")
AUTHORIZED_UUID = os.environ.get("AUTHORIZED_ACCOUNT_UUID", "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx")

def is_authorized(envelope: dict) -> bool:
    source_number = envelope.get("sourceNumber")
    source_uuid = envelope.get("sourceUuid")
    
    # Reject any message that did not originate from my personal account
    if source_number != AUTHORIZED_NUMBER and source_uuid != AUTHORIZED_UUID:
        return False
        
    # Strictly reject group messages to prevent prompt injection from group members
    data_message = envelope.get("dataMessage", {})
    if "groupInfo" in data_message:
        return False
        
    return True

This ensures that the bot only responds to messages sent to “Note to Self” from my authenticated device. Any message from external contacts or shared groups is silently ignored.

Systemd User Service (signal-bot.service)

The gateway runs as a persistent systemd --user service with auto-restart on network failure:

[Unit]
Description=Signal Gateway Daemon for Antigravity AI Assistant
After=network.target sound.target

[Service]
Type=simple
ExecStart=/usr/bin/python3 %h/scripts/signal_bot.py
Restart=always
RestartSec=5s
Environment=PYTHONUNBUFFERED=1
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=default.target

To manage the daemon:

# Check running status and recent logs
systemctl --user status signal-bot.service

# Restart daemon after updating scripts
systemctl --user restart signal-bot.service

# View real-time logs
journalctl --user -u signal-bot.service -f

8. Lessons Learned & Key Takeaways

Building a responsive mobile interface for a local AI agent taught several foundational engineering lessons:

1. Do Not Make LLMs Do Deterministic Work: Using an agent to inspect a sensor or ping a server is an anti-pattern. Intercept routine tasks with clean code paths; reserve the LLM for reasoning, synthesis, and novel problem-solving.
2. Visual Feedback Trumps Raw Speed: Even a 6-second delay feels natural if a typing indicator appears within 100 milliseconds. Without that indicator, users assume the system is dead.
3. Session Windows Must Match Human Behavior: Retaining a 2-hour conversational window strikes the ideal balance—long enough to resume a thought after a short walk, yet short enough to avoid carrying obsolete context into the next day.
4. Cap Your Agent Timeouts: Autonomous agents with tool access can get caught in recursive debugging loops. Enforcing a strict per-turn execution bound (e.g., 5 minutes) and graceful fallback is vital for headless reliability.

By combining deterministic fast-paths, real-time JSON-RPC typing indicators, and session continuity, my local AI assistant on Signal went from feeling like a sluggish batch script to a delightful, instant pocket companion.

Useful Links & References

• Signal-CLI Project: github.com/AsahiLinux/signal-cli
• Google Antigravity CLI: agy documentation and local integration rules