The Touch Grass Manifesto: Building AI Apps That Force Developers Outside

// Core invariant: zero screen dependencies in the runtime
// This compiles away if any UI framework is linked
#[cfg(target_os = "ios")]
compile_error!("Touch Grass apps cannot link UIKit. Use AudioToolbox + CoreMotion only.");

#[cfg(target_os = "android")]
compile_error!("Touch Grass apps cannot link Android View system. Use AudioTrack + SensorManager only.");

// Swift: Minimal llama.cpp wrapper for voice streaming
// TouchGrassInference.swift
import Foundation
import llama

final class TouchGrassInference {
    private var ctx: OpaquePointer?
    private var batch: llama_batch
    private let nCtx = 8192
    private let nThreads = ProcessInfo.processInfo.activeProcessorCount - 1

    init(modelPath: String) throws {
        var params = llama_context_default_params()
        params.n_ctx = Int32(nCtx)
        params.n_threads = Int32(nThreads)
        params.n_threads_batch = Int32(nThreads)
        params.offload_kqv = true  // Metal/GPU offload
        params.flash_attn = true
        
        ctx = llama_init_from_file(modelPath, params)
        guard ctx != nil else { throw InferenceError.modelLoadFailed }
        
        batch = llama_batch_init(512, 0, 1)
    }

    // Streaming callback for token-by-token TTS handoff
    func generate(
        prompt: String,
        onToken: @escaping (String) -> Bool  // return false to stop
    ) throws {
        let tokens = tokenize(prompt)
        batch.n_tokens = Int32(tokens.count)
        for (i, tok) in tokens.enumerated() {
            batch.token[i] = tok
            batch.pos[i] = Int32(i)
            batch.n_seq_id[i][0] = 0
            batch.logits[i] = (i == tokens.count - 1) ? 1 : 0
        }
        
        if llama_decode(ctx!, batch) != 0 { throw InferenceError.decodeFailed }
        
        var nPast = tokens.count
        while true {
            let logits = llama_get_logits_ith(ctx!, Int32(batch.n_tokens - 1))
            let nextTok = sampleTopP(logits, p: 0.9, temp: 0.7)
            
            if llama_token_is_eog(ctx!, nextTok) { break }
            
            let piece = String(cString: llama_token_to_piece(ctx!, nextTok))
            if !onToken(piece) { break }  // TTS or user interrupted
            
            batch.n_tokens = 1
            batch.token[0] = nextTok
            batch.pos[0] = Int32(nPast)
            batch.logits[0] = 1
            nPast += 1
            
            if llama_decode(ctx!, batch) != 0 { throw InferenceError.decodeFailed }
            
            if nPast >= nCtx { throw InferenceError.contextFull }
        }
    }

    deinit { if let ctx { llama_free(ctx) } }
}

// ModelManager.swift
enum ModelProfile: String, CaseIterable {
    case coder = "llama-3.2-3b-coder-q4"
    case storyteller = "gemma-2-2b-story-q4"
    case navigator = "phi-3.5-mini-nav-q4"
    
    var hapticPattern: CHHapticPattern { /* distinct per model */ }
    var earcon: String { /* unique 200ms audio ID */ }
}

final class ModelManager {
    private var current: TouchGrassInference?
    private let audio: AudioEngine
    private let haptics: HapticEngine
    
    func switchTo(_ profile: ModelProfile) async throws {
        // Play earcon *before* unload so user hears intent
        try await audio.playEarcon(profile.earcon)
        try await haptics.play(profile.hapticPattern)
        
        current = try TouchGrassInference(modelPath: profile.path)
        
        // Confirmation: speak model name via TTS
        try await audio.speak("Switched to \(profile.rawValue)")
    }
}

┌─────────────┐   ┌──────────────┐   ┌─────────────┐   ┌──────────────┐
│  Mic Array  │──▶│  VAD + AEC   │──▶│  ASR Stream │──▶│  Intent/Slot │
│  (beamform) │   │  (RNNoise)   │   │  (Whisper)  │   │  (LLM)       │
└─────────────┘   └──────────────┘   └─────────────┘   └──────────────┘
       ▲                                       │                   │
       │                    ┌──────────────────┘                   ▼
       │                    ▼                          ┌─────────────────┐
       │             ┌──────────────┐                  │  TTS (Piper/    │
       └─────────────│  Barge-in    │◀─────────────────│  StyleTTS2)     │
                     │  Detection   │   token stream   └─────────────────┘
                     └──────────────┘                        │
                            │                                ▼
                            ▼                       ┌─────────────────┐
                     ┌──────────────┐                │  Audio Mixer    │
                     │  Turn Manager│                │  (spatialized)  │
                     └──────────────┘                └─────────────────┘

# faster-whisper streaming wrapper
# asr_stream.py
from faster_whisper import WhisperModel
import numpy as np
import queue
import threading

class StreamingASR:
    def __init__(self, model_size="tiny.en", device="cpu", compute_type="int8"):
        self.model = WhisperModel(model_size, device=device, compute_type=compute_type)
        self.audio_queue = queue.Queue()
        self.result_queue = queue.Queue()
        self.running = False
        self.buffer = np.array([], dtype=np.float32)
        self.sample_rate = 16000
        self.chunk_duration = 0.5  # 500ms chunks
        
    def start(self):
        self.running = True
        threading.Thread(target=self._process_loop, daemon=True).start()
    
    def push_audio(self, chunk: np.ndarray):
        self.audio_queue.put(chunk)
    
    def _process_loop(self):
        while self.running:
            try:
                chunk = self.audio_queue.get(timeout=0.1)
                self.buffer = np.concatenate([self.buffer, chunk])
                
                # Process when we have enough context
                if len(self.buffer) >= self.sample_rate * 2:  # 2s window
                    segments, _ = self.model.transcribe(
                        self.buffer[-self.sample_rate*4:],  # last 4s
                        language="en",
                        vad_filter=True,
                        vad_parameters=dict(min_silence_duration_ms=500)
                    )
                    text = " ".join(s.text for s in segments).strip()
                    if text:
                        self.result_queue.put(text)
                    # Keep last 1s for context overlap
                    self.buffer = self.buffer[-self.sample_rate:]
            except queue.Empty:
                continue

// BargeInManager.swift
final class BargeInManager {
    private let vad: VoiceActivityDetector  // Silero VAD, 1.5ms/frame
    private let tts: StreamingTTS
    private let asr: StreamingASR
    private var isSpeaking = false
    
    func onTTSStart() { isSpeaking = true }
    
    func onTTSChunk(_ audio: Data) {
        // Feed TTS output back to VAD for echo cancellation reference
        vad.referenceSignal(audio)
    }
    
    func onMicAudio(_ audio: Data) -> BargeInDecision {
        let voiceProb = vad.process(audio)
        
        if isSpeaking && voiceProb > 0.85 {
            // User speaking over TTS → barge in
            tts.stopImmediately()
            isSpeaking = false
            asr.reset()  // Clear any partial hypothesis
            return .bargeIn
        }
        
        if !isSpeaking && voiceProb > 0.6 {
            return .startListening
        }
        
        return .continue
    }
}

// ConfirmationStrategy.swift
enum ConfirmationLevel {
    case none      // Read-only, low risk
    case implicit  // "I'll save that note" → 3s undo window
    case explicit  // "Delete all notes? Say 'yes delete' to confirm"
    case multiModal // "Say 'yes' AND tap phone twice" (for destructive)
}

func confirmationLevel(for action: Action) -> ConfirmationLevel {
    switch action {
    case .createNote: return .implicit
    case .sendMessage: return .explicit
    case .deleteAll: return .multiModal
    case .runCode: return .explicit  // Code execution = explicit
    }
}

// WalkContext.swift
import CoreMotion
import CoreLocation

struct WalkContext {
    let pace: Pace          // .stroll / .walk / .brisk / .run
    let terrain: Terrain    // .pavement / .trail / .stairs / .unknown
    let environment: Env    // .quiet / .street / .park / .transit
    let location: CLLocation?
    let timeOfDay: TimePhase
    let weather: Weather?
    let heartRate: Double?  // if HealthKit authorized
    
    var narrativeTempo: NarrativeTempo {
        switch (pace, terrain) {
        case (.stroll, .park): return .reflective
        case (.brisk, .pavement): return .urgent
        case (.walk, .trail): return .exploratory
        case (.run, _): return .action
        default: return .neutral
        }
    }
}

final class WalkContextEngine {
    private let motion = CMMotionManager()
    private let location = CLLocationManager()
    private let altimeter = CMAltimeter()
    private var contextContinuation: AsyncStream<WalkContext>.Continuation?
    
    var contextStream: AsyncStream<WalkContext> {
        AsyncStream { continuation in
            self.contextContinuation = continuation
            self.startUpdates()
        }
    }
    
    private func startUpdates() {
        motion.startDeviceMotionUpdates(to: .main) { [weak self] data, _ in
            guard let data, let self else { return }
            let pace = self.classifyPace(data)
            let terrain = self.classifyTerrain(data)
            // ... fuse with location, weather, time
            let ctx = WalkContext(...)
            self.contextContinuation?.yield(ctx)
        }
    }
    
    private func classifyPace(_ data: CMDeviceMotion) -> Pace {
        // Vertical oscillation + step frequency from userAcceleration
        let vertical = data.userAcceleration.z
        let freq = self.stepFrequency(from: data)
        
        switch (freq, abs(vertical)) {
        case (0..<1.5, _): return .stroll
        case (1.5..<2.0, 0..<0.15): return .walk
        case (1.5..<2.0, _): return .brisk
        case (2.0..., _): return .run
        default: return .unknown
        }
    }
}

// story_graph.json - loaded at startup, no screen needed
{

{
  "nodes": {
    "start": {
      "text": "The trailhead is empty. Your boots hit dirt. What's the first thing you notice?",
      "choices": [
        { "label": "The smell of pine", "next": "pine", "requires": {} },
        { "label": "Birdsong overhead", "next": "birds", "requires": {} },
        { "label": "The weight of your pack", "next": "pack", "requires": {} }
      ]
    },
    "pine": {
      "text": "Pine needles cushion each step. The air tastes like resin and rain. A deer watches from thirty meters.",
      "choices": [
        { "label": "Freeze. Watch it leave.", "next": "deer_gone", "requires": {} },
        { "label": "Whistle low. See if it approaches.", "next": "deer_curious", "requires": { "has_whistle": true } }
      ]
    },
    "birds": {
      "text": "A Steller's jay scolds you from a cedar. Two chickadees mob it. The canopy is alive with argument.",
      "choices": [
        { "label": "Identify the jay's alarm call.", "next": "bird_id", "requires": { "skill": "birding" } },
        { "label": "Walk on. They're not your problem.", "next": "ridge", "requires": {} }
      ]
    },
    "pack": {
      "text": "Twenty kilos. Water, shelter, the satellite messenger your partner insisted on. Every gram earned its place.",
      "choices": [
        { "label": "Adjust the hip belt. Keep moving.", "next": "ridge", "requires": {} },
        { "label": "Dump the camp chair. Save six hundred grams.", "next": "lighter", "requires": { "has_chair": true } }
      ]
    },
    "ridge": {
      "text": "The trail breaks onto a ridge. Valley spreads below—river silver, forest green, the highway a gray scar.",
      "choices": [
        { "label": "Sit. Eat the apple you packed.", "next": "apple", "requires": { "has_apple": true } },
        { "label": "Pull out the messenger. Check in.", "next": "checkin", "requires": { "has_messenger": true } },
        { "label": "Just breathe. No screens.", "next": "breathe", "requires": {} }
      ]
    },
    "breathe": {
      "text": "Wind carries wet earth and snowmelt. Your shoulders drop. This is why you came.",
      "is_ending": true,
      "ending_type": "pure"
    },
    "checkin": {
      "text": "Three taps. 'All good. On ridge. Back by dark.' The satellite swallows it. Peace of mind, delivered.",
      "is_ending": true,
      "ending_type": "connected"
    }
  },
  "entry": "start",
  "meta": {
    "version": "1.0",
    "estimated_walk_minutes": 45,
    "difficulty": "moderate"
  }
}

# story_engine.py - zero dependencies, ~80 lines
import json
from dataclasses import dataclass
from typing import Optional

@dataclass
class WalkContext:
    step_count: int
    elevation_gain_m: float
    heart_rate_bpm: Optional[int]
    inventory: set[str]
    skills: set[str]
    time_of_day: str  # "dawn" | "day" | "dusk" | "night"
    weather: str      # "clear" | "cloudy" | "rain" | "storm"

@dataclass
class Choice:
    label: str
    next_node: str
    requires: dict

@dataclass
class Node:
    text: str
    choices: list[Choice]
    is_ending: bool = False
    ending_type: str = ""

def load_graph(path: str) -> dict[str, Node]:
    with open(path) as f:
        raw = json.load(f)
    nodes = {}
    for nid, nd in raw["nodes"].items():
        nodes[nid] = Node(
            text=nd["text"],
            choices=[Choice(**c) for c in nd.get("choices", [])],
            is_ending=nd.get("is_ending", False),
            ending_type=nd.get("ending_type", "")
        )
    return nodes

def available_choices(node: Node, ctx: WalkContext) -> list[Choice]:
    """Filter choices by context. Pure, testable, no I/O."""
    def check(req: dict) -> bool:
        for k, v in req.items():
            if k == "has_whistle" and v and "whistle" not in ctx.inventory:
                return False
            if k == "has_chair" and v and "camp_chair" not in ctx.inventory:
                return False
            if k == "has_apple" and v and "apple" not in ctx.inventory:
                return False
            if k == "has_messenger" and v and "sat_messenger" not in ctx.inventory:
                return False
            if k == "skill" and v not in ctx.skills:
                return False
        return True
    return [c for c in node.choices if check(c.requires)]

def render_node(node: Node, ctx: WalkContext) -> str:
    lines = [node.text]
    for i, ch in enumerate(available_choices(node, ctx), 1):
        lines.append(f"  {i}. {ch.label}")
    return "\n".join(lines)

# build_audio.sh - runs in CI, commits artifacts
#!/usr/bin/env bash
set -euo pipefail

GRAPH="story_graph.json"
OUT_DIR="assets/audio"
VOICE="en_US-lessac-medium"  # Piper voice model

mkdir -p "$OUT_DIR"

# Extract all unique node texts
jq -r '.nodes[].text' "$GRAPH" | sort -u | while IFS= read -r text; do
  # Stable filename: sha256 of text
  fname=$(echo -n "$text" | sha256sum | cut -c1-16)
  out="$OUT_DIR/$fname.opus"
  [[ -f "$out" ]] && continue  # idempotent
  echo "$text" | piper --model "$VOICE" --output_raw | \
    opusenc --bitrate 24 --raw --raw-channels 1 --raw-rate 22050 - "$out"
done

# Generate manifest for the app
jq -r '.nodes | to_entries[] | "\(.key) \(.value.text)"' "$GRAPH" | \
while read -r nid text; do
  fname=$(echo -n "$text" | sha256sum | cut -c1-16)
  echo "{\"node\":\"$nid\",\"audio\":\"$fname.opus\"}"
done > "$OUT_DIR/manifest.jsonl"

// story_graph.json snippet
{
  "nodes": {
    "checkin": {
      "text": "Three taps. 'All good. {{step_count}} steps. {{elevation}} meters up. Back by dark.'",
      "audio_template": "checkin_base.opus",
      "dynamic_slots": ["step_count", "elevation"]
    }
  }
}

# context_builder.py - runs on phone, updates every 5s
from dataclasses import dataclass
from enum import Enum
import time

class Weather(Enum):
    CLEAR = "clear"
    CLOUDY = "cloudy"
    RAIN = "rain"
    STORM = "storm"

@dataclass
class WalkContext:
    step_count: int
    elevation_gain_m: float
    heart_rate_bpm: int | None
    inventory: set[str]
    skills: set[str]
    time_of_day: str
    weather: Weather

class ContextBuilder:
    # ponytail: thresholds tuned for iOS/Android sensor noise profiles.
    # Upgrade path: per-device calibration on first run.
    PACE_WINDOW_SEC = 30
    HR_MIN_VALID = 40
    HR_MAX_VALID = 220
    ELEVATION_SMOOTH_ALPHA = 0.3

    def __init__(self):
        self._step_timestamps: list[float] = []
        self._elevation_ewma: float | None = None
        self._last_pressure_hpa: float | None = None

    def update_steps(self, new_steps: int, timestamp: float) -> int:
        self._step_timestamps.append(timestamp)
        # Drop old
        cutoff = timestamp - self.PACE_WINDOW_SEC
        self._step_timestamps = [t for t in self._step_timestamps if t > cutoff]
        return new_steps

    def current_pace_spm(self) -> float:
        if len(self._step_timestamps) < 2:
            return 0.0
        dt = self._step_timestamps[-1] - self._step_timestamps[0]
        return len(self._step_timestamps) / dt * 60  # steps/min

    def update_elevation(self, pressure_hpa: float) -> float:
        """Barometric altitude. No GPS vertical—too noisy."""
        if self._last_pressure_hpa is None:
            self._last_pressure_hpa = pressure_hpa
            self._elevation_ewma = 0.0
            return 0.0
        # Hypsometric formula, simplified
        delta_h = 8.3 * (self._last_pressure_hpa - pressure_hpa)  # meters per hPa approx
        self._elevation_ewma = (
            self.ELEVATION_SMOOTH_ALPHA * delta_h +
            (1 - self.ELEVATION_SMOOTH_ALPHA) * self._elevation_ewma
        )
        self._last_pressure_hpa = pressure_hpa
        return max(0.0, self._elevation_ewma)

    def update_heart_rate(self, bpm: int | None) -> int | None:
        if bpm is None:
            return None
        if not (self.HR_MIN_VALID <= bpm <= self.HR_MAX_VALID):
            return None  # discard artifact
        return bpm

    def infer_weather(self, pressure_hpa: float, humidity: float, temp_c: float) -> Weather:
        # ponytail: naive heuristic. Ceiling: no microclimate awareness.
        # Upgrade: on-device TinyML model (TensorFlow Lite, <100KB).
        if pressure_hpa < 990 and humidity > 80:
            return Weather.STORM
        if pressure_hpa < 1000 and humidity > 70:
            return Weather.RAIN
        if humidity > 60:
            return Weather.CLOUDY
        return Weather.CLEAR

    def time_of_day(self) -> str:
        h = time.localtime().tm_hour
        if 5 <= h < 8: return "dawn"
        if 8 <= h < 18: return "day"
        if 18 <= h < 21: return "dusk"
        return "night"

// iOS: AudioSession + Background Modes = "audio" + "location updates"
// Android: Foreground Service (mediaPlayback) + PARTIAL_WAKE_LOCK

class AudioSessionManager {
    func configure() throws {
        let session = AVAudioSession.sharedInstance()
        try session.setCategory(.playback, mode: .spokenAudio, options: [.duckOthers, .mixWithOthers])
        try session.setActive(true)
        // Lock screen controls appear automatically. No UI code needed.
    }
}

# test_story_engine.py
import hypothesis.strategies as st
from hypothesis import given, settings
from story_engine import load_graph, available_choices, WalkContext

GRAPH = load_graph("story_graph.json")

@given(
    step_count=st.integers(0, 50000),
    elevation=st.floats(0, 3000),
    hr=st.one_of(st.none(), st.integers(30, 250)),
    inventory=st.sets(st.sampled_from(["whistle", "camp_chair", "apple", "sat_messenger"])),
    skills=st.sets(st.sampled_from(["birding", "navigation", "first_aid"])),
    time_of_day=st.sampled_from(["dawn", "day", "dusk", "night"]),
    weather=st.sampled_from(["clear", "cloudy", "rain", "storm"]),
)
@settings(max_examples=500, deadline=None)
def test_no_crashes_on_valid_context(step_count, elevation, hr, inventory, skills, time_of_day, weather):
    ctx = WalkContext(step_count, elevation, hr, inventory, skills, time_of_day, weather)
    for node in GRAPH.values():
        # Should never raise, even with nonsense context
        _ = available_choices(node, ctx)

@given(nid=st.sampled_from(list(GRAPH.keys())))
def test_every_node_reachable_from_start(nid):
    # BFS from entry
    from collections import deque
    visited = set()
    q = deque([GRAPH["start"]])
    while q:
        node = q.popleft()
        if id(node) in visited: continue
        visited.add(id(node))
        for ch in node.choices:
            if ch.next_node == nid:
                return  # found
            q.append(GRAPH[ch.next_node])
    assert False, f"Node {nid} unreachable from start"

def test_no_dead_ends_except_endings():
    for nid, node in GRAPH.items():
        if node.is_ending: continue
        # At least one choice with empty requires (always available)
        assert any(len(c.requires) == 0 for c in node.choices), f"{nid}: all choices gated"

trail/
├── Cargo.toml           # or pyproject.toml / package.json / go.mod
├── src/
│   ├── main.rs          # 200 lines: init sensors → loop → render → play
│   ├── story_engine.rs  # the pure logic above
│   ├── context.rs       # sensor readers (platform-specific, ~150 lines each)
│   └── audio.rs         # playback + manifest lookup
├── assets/
│   ├── story_graph.json
│   └── audio/           # 50 MB .opus + manifest.jsonl (git-lfs)
└── .github/workflows/
    ├── build.yml        # compiles, runs tests, generates audio
    └── release.yml      # codesign → notarize → upload to TestFlight / Play Console

-- The only query that matters
SELECT COUNT(*)
FROM walks
WHERE phone_screen_on_seconds < 30
  AND duration_minutes BETWEEN 20 AND 180
  AND ended_at_ridge = true;