ALMA PRESENCE — CONSOLIDATED PROTOTYPE BACKEND SOURCE Generated from the tested reference implementation. To reconstruct the project, save each section under the path shown after FILE. This source is a prototype and is not a certified medical, emergency or production security system. ======================================================================================== FILE: README.md ======================================================================================== # ALMA Presence functional-alpha backend This package is a runnable reference implementation for the highest-value and highest-risk backend behaviours in the ALMA Presence tabletop prototype: - source-attributed personal facts that remain unusable until verified; - reminders persisted in SQLite so they survive process restart and internet loss; - trusted contacts and a resident-specific smart-home allowlist; - deterministic server-side action classification; - short-lived confirmation tokens bound to the exact normalized action; - closed, typed tool adapters with idempotent action proposals and receipts; - hardware privacy, network and motor-health event state; and - an append-only resident audit trail. The language model is not a privileged service. It may produce an action proposal, but it receives no telephone credentials, smart-home token, motor interface, arbitrary URL tool or shell access. ## What is intentionally mocked AlmaTalk, Home Assistant and information adapters return realistic demo receipts but do not place real calls, send messages, create bookings or control devices. `ALMA_EXECUTION_MODE` is hard-limited to `mock` in this package. Connect production adapters only after their schemas, credentials, idempotency, timeout behaviour, receipts, disclosure rules and failure modes have been independently reviewed. Do not add a generic HTTP-request or shell adapter. ## Quick start with Docker 1. Copy `.env.example` to `.env`. 2. Replace `ALMA_API_KEY` and `ALMA_CONFIRMATION_SECRET` with random secrets. 3. Run `docker compose up --build`. 4. Open `http://127.0.0.1:8080/docs` for the generated API interface. 5. In another terminal, run `ALMA_API_KEY= python scripts/demo.py`. The default local database is stored in the Docker volume `alma-data`. ## Run without Docker Requires Python 3.12 or newer. ```bash python -m venv .venv source .venv/bin/activate pip install -r requirements.txt cp .env.example .env uvicorn app.main:app --reload --port 8080 --env-file .env ``` Set the environment variables in `.env` through your normal process manager or secret store. The API reads environment variables directly; it does not load credentials from code or the database. ## Test the safety core The policy and service tests use only the Python standard library, so they can run before the web dependencies are installed: ```bash PYTHONPATH=. python -m unittest discover -s tests -v ``` The suite verifies that: - non-allowlisted home entities are blocked; - trusted messages require confirmation; - door unlocking and financial transfer remain blocked; - confirmation tokens fail when the action, payload or expiry changes; - proposed personal facts cannot be retrieved as truth; - a verified, in-date, purpose-permitted fact can be retrieved with its source; - low-risk local actions produce receipts; and - reminder acknowledgement is recorded without claiming medication adherence. ## API workflow Every protected route requires `X-ALMA-API-Key`. Actor-changing operations also accept `X-ALMA-Actor` with one of `resident`, `caregiver`, `device`, `hardware` or `support`. The API-key scheme is appropriate for an isolated functional alpha only; replace it with device mTLS and real resident/caregiver identity before any external pilot. ### 1. Configure the resident context - `POST /v1/residents` - `POST /v1/residents/{id}/contacts` - `PUT /v1/residents/{id}/home-entities/{entity_id}` - `POST /v1/residents/{id}/devices` Contact destinations are stored as opaque provider references. Avoid placing raw provider credentials, OAuth tokens or unnecessary phone numbers in this database. ### 2. Build verified memory - `POST /v1/residents/{id}/facts` creates a `proposed` fact. - `POST /v1/facts/{id}/verify` requires a resident or caregiver actor. - `GET /v1/residents/{id}/facts/query` returns only facts that are verified, valid at the current time and permitted for the requested purpose. Conversation-derived memories must enter as `proposed`. The model cannot set `verification_status`, and a vector similarity result is never enough to turn an inference into personal truth. ### 3. Store offline reminders - `POST /v1/residents/{id}/reminders` - `GET /v1/residents/{id}/reminders/due` - `POST /v1/reminders/{id}/acknowledge` The reference alpha uses explicit timezone-aware timestamps. Production recurrence handling should be added with a reviewed iCalendar/RRULE library and tested across daylight-saving transitions. Acknowledgement means only that the resident responded to the reminder; it is not evidence of medication ingestion. ### 4. Propose and confirm actions - `POST /v1/actions/proposals` - `POST /v1/actions/{id}/confirm` - `GET /v1/actions/{id}/receipt` The action state machine is: ```text proposed -> policy_checked -> confirmation_required -> confirmed -> executing -> succeeded | failed | blocked | expired ``` `request_id` is unique. Replaying the same proposal returns the stored action rather than executing twice. A confirmation token binds the resident, device, action type, normalized parameters and short expiry. Changing recipient, message, time, price or any other field invalidates confirmation. Current alpha policy: | Risk class | Examples | Result | |---|---|---| | 0 | Read approved information | Allow | | 1 | Allowlisted light or media control | Allow and write receipt | | 2 | Call or message an authorized trusted contact | Exact confirmation | | 3 | Booking with exact provider, time, price and cancellation terms | Exact confirmation | | 4 | Exterior door, garage or water actuation | Block in alpha | | 5 | Financial transfer, purchase, stove/oven, alarm disable, covert camera, medication dispensing | Always block | Unknown action types and invalid parameter envelopes are blocked, not guessed. ## Local/cloud production split Keep these capabilities on the edge device and available without internet: - hardware privacy controls, wake word, voice activity and core avatar state; - basic ASR/TTS, captions and interruption; - verified fact/routine cache and reminder scheduler; - proactivity budget, quiet hours and cooldowns; - allowlisted Home Assistant light/media control; - presence, face location and user-initiated common-object assistance; and - motor safety IPC plus static-screen fallback. Network-dependent capabilities may include caregiver identity and remote configuration, encrypted backup/sync, AlmaTalk PSTN/WebRTC and messaging, external calendar OAuth, booking workflows, optional advanced models, fleet health and signed update metadata. Raw room audio/video, arbitrary transcripts, face templates and local smart-home credentials should not upload by default. A cloud vision escalation should require a user-initiated frame, a clear network indicator and a no-retention provider agreement. ## Repository path from alpha to production The functional package is intentionally compact. The recommended funded-alpha repository expands into: ```text apps/cloud-api modular cloud API apps/cloud-worker durable calls/messages/bookings jobs apps/edge-orchestrator local voice, memory and policy apps/edge-supervisor process, update and hardware health apps/device-ui kiosk avatar and captions apps/care-web caregiver onboarding and audit packages/contracts generated TypeScript/Python clients packages/policy shared rules and golden tests packages/almatalk-adapter packages/home-assistant-adapter infra/terraform deploy/edge-compose tests/device-simulator tests/adversarial ``` For the alpha, keep the cloud side as one modular API, one durable worker, PostgreSQL, object storage and a secrets manager in a Canadian region. Do not start with a microservice fleet. ## Before a home pilot This backend is not production-ready or clinically validated. Complete at least the following first: - privacy impact assessment and consent/capacity policy reviewed for Canada and Quebec; - device mTLS, secure key storage, encrypted local database, signed updates, rollback and device revocation; - penetration testing, dependency review, retention/deletion tests and support access controls; - purpose-based authorization for resident, family, professional caregiver and time-limited support roles; - AlmaTalk provider review, AI-caller disclosure and Canadian VoIP 9-1-1 onboarding with a verified civic address; - independent microphone disconnect and camera shutter sensing; - motor watchdog, hard stops, obstruction/current limits, anti-tip, pinch, thermal and electrical testing; - accessibility testing with older adults and caregivers; and - a claims review that avoids treatment, diagnosis, fall detection, medication-adherence, guaranteed safety or emergency-response claims. The initial product should be positioned as a communication, accessibility, scheduling, general-wellness and smart-home companion until evidence and professional review support anything further. ======================================================================================== FILE: requirements.txt ======================================================================================== fastapi==0.116.1 uvicorn[standard]==0.35.0 pydantic==2.11.7 python-dotenv==1.1.1 ======================================================================================== FILE: Dockerfile ======================================================================================== FROM python:3.12-slim ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app ./app COPY contracts ./contracts RUN mkdir -p /app/data && chown -R nobody:nogroup /app USER nobody EXPOSE 8080 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"] ======================================================================================== FILE: docker-compose.yml ======================================================================================== services: alma-presence-api: build: . ports: - "8080:8080" env_file: - .env volumes: - alma-data:/app/data read_only: true tmpfs: - /tmp security_opt: - no-new-privileges:true restart: unless-stopped volumes: alma-data: ======================================================================================== FILE: app/__init__.py ======================================================================================== """ALMA Presence functional-alpha backend.""" ======================================================================================== FILE: app/settings.py ======================================================================================== from __future__ import annotations from dataclasses import dataclass import os @dataclass(frozen=True) class Settings: database_path: str api_key: str confirmation_secret: str execution_mode: str policy_version: str confirmation_ttl_seconds: int def load_settings() -> Settings: ttl = int(os.getenv("ALMA_CONFIRMATION_TTL_SECONDS", "120")) if ttl < 30 or ttl > 600: raise ValueError("ALMA_CONFIRMATION_TTL_SECONDS must be between 30 and 600") execution_mode = os.getenv("ALMA_EXECUTION_MODE", "mock").strip().lower() if execution_mode != "mock": raise ValueError( "This reference backend intentionally supports only ALMA_EXECUTION_MODE=mock. " "Connect reviewed provider adapters before enabling real-world actions." ) return Settings( database_path=os.getenv("ALMA_DATABASE_PATH", "./data/alma_presence.db"), api_key=os.getenv("ALMA_API_KEY", "local-demo-key"), confirmation_secret=os.getenv( "ALMA_CONFIRMATION_SECRET", "local-demo-secret-change-before-sharing", ), execution_mode=execution_mode, policy_version=os.getenv("ALMA_POLICY_VERSION", "alpha-1"), confirmation_ttl_seconds=ttl, ) ======================================================================================== FILE: app/database.py ======================================================================================== from __future__ import annotations from contextlib import contextmanager from pathlib import Path import sqlite3 from typing import Iterator SCHEMA = """ CREATE TABLE IF NOT EXISTS residents ( id TEXT PRIMARY KEY, preferred_name TEXT NOT NULL, locale TEXT NOT NULL DEFAULT 'en-CA', timezone TEXT NOT NULL DEFAULT 'America/Toronto', quiet_start TEXT NOT NULL DEFAULT '21:00', quiet_end TEXT NOT NULL DEFAULT '07:00', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS trusted_contacts ( id TEXT PRIMARY KEY, resident_id TEXT NOT NULL REFERENCES residents(id) ON DELETE CASCADE, display_name TEXT NOT NULL, relationship TEXT NOT NULL, destination_ref TEXT NOT NULL, call_window_start TEXT NOT NULL DEFAULT '07:00', call_window_end TEXT NOT NULL DEFAULT '21:00', allow_calls INTEGER NOT NULL DEFAULT 1 CHECK (allow_calls IN (0, 1)), allow_messages INTEGER NOT NULL DEFAULT 0 CHECK (allow_messages IN (0, 1)), disclosure_json TEXT NOT NULL DEFAULT '[]', created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS contacts_resident_idx ON trusted_contacts(resident_id, display_name); CREATE TABLE IF NOT EXISTS memory_facts ( id TEXT PRIMARY KEY, resident_id TEXT NOT NULL REFERENCES residents(id) ON DELETE CASCADE, subject TEXT NOT NULL, predicate TEXT NOT NULL, value_json TEXT NOT NULL, source_kind TEXT NOT NULL, source_reference TEXT NOT NULL, created_by TEXT NOT NULL, verification_status TEXT NOT NULL DEFAULT 'proposed' CHECK (verification_status IN ('proposed', 'verified', 'rejected', 'expired')), verified_by TEXT, verified_at TEXT, confidence REAL NOT NULL DEFAULT 1.0 CHECK (confidence >= 0 AND confidence <= 1), sensitivity TEXT NOT NULL DEFAULT 'personal' CHECK (sensitivity IN ('public', 'personal', 'sensitive', 'restricted')), allowed_purposes_json TEXT NOT NULL DEFAULT '["speak"]', valid_from TEXT, valid_until TEXT, correction_of_id TEXT REFERENCES memory_facts(id), created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS facts_lookup_idx ON memory_facts(resident_id, predicate, verification_status, valid_until); CREATE TABLE IF NOT EXISTS reminders ( id TEXT PRIMARY KEY, resident_id TEXT NOT NULL REFERENCES residents(id) ON DELETE CASCADE, title TEXT NOT NULL, due_at TEXT NOT NULL, timezone TEXT NOT NULL, recurrence_rule TEXT, source_reference TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'scheduled' CHECK (status IN ('scheduled', 'delivered', 'acknowledged', 'snoozed', 'cancelled')), acknowledgement_note TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS reminders_due_idx ON reminders(status, due_at); CREATE TABLE IF NOT EXISTS devices ( id TEXT PRIMARY KEY, resident_id TEXT NOT NULL REFERENCES residents(id) ON DELETE CASCADE, display_name TEXT NOT NULL, privacy_shutter_closed INTEGER NOT NULL DEFAULT 0 CHECK (privacy_shutter_closed IN (0, 1)), microphone_disconnected INTEGER NOT NULL DEFAULT 0 CHECK (microphone_disconnected IN (0, 1)), motor_healthy INTEGER NOT NULL DEFAULT 1 CHECK (motor_healthy IN (0, 1)), network_online INTEGER NOT NULL DEFAULT 1 CHECK (network_online IN (0, 1)), last_seen_at TEXT, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS home_entity_allowlist ( id TEXT PRIMARY KEY, resident_id TEXT NOT NULL REFERENCES residents(id) ON DELETE CASCADE, entity_id TEXT NOT NULL, allowed_actions_json TEXT NOT NULL, value_constraints_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL, UNIQUE(resident_id, entity_id) ); CREATE TABLE IF NOT EXISTS action_proposals ( id TEXT PRIMARY KEY, request_id TEXT NOT NULL UNIQUE, resident_id TEXT NOT NULL REFERENCES residents(id) ON DELETE CASCADE, device_id TEXT REFERENCES devices(id) ON DELETE SET NULL, actor_kind TEXT NOT NULL, action_type TEXT NOT NULL, normalized_params_json TEXT NOT NULL, payload_hash TEXT NOT NULL, risk_class INTEGER NOT NULL, state TEXT NOT NULL, decision TEXT NOT NULL, reason_codes_json TEXT NOT NULL, policy_version TEXT NOT NULL, confirmation_token_hash TEXT, confirmation_expires_at TEXT, confirmed_at TEXT, receipt_json TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS actions_resident_idx ON action_proposals(resident_id, created_at); CREATE TABLE IF NOT EXISTS audit_events ( id TEXT PRIMARY KEY, resident_id TEXT, actor_kind TEXT NOT NULL, event_type TEXT NOT NULL, entity_type TEXT NOT NULL, entity_id TEXT NOT NULL, details_json TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS audit_resident_idx ON audit_events(resident_id, created_at); CREATE TRIGGER IF NOT EXISTS audit_no_update BEFORE UPDATE ON audit_events BEGIN SELECT RAISE(ABORT, 'audit events are append-only'); END; CREATE TRIGGER IF NOT EXISTS audit_no_delete BEFORE DELETE ON audit_events BEGIN SELECT RAISE(ABORT, 'audit events are append-only'); END; """ class ManagedConnection(sqlite3.Connection): """SQLite connection that closes when used as a context manager.""" def __exit__(self, exc_type, exc_value, traceback): result = super().__exit__(exc_type, exc_value, traceback) self.close() return result class Database: def __init__(self, path: str): self.path = path if path != ":memory:": Path(path).expanduser().resolve().parent.mkdir(parents=True, exist_ok=True) def connect(self) -> sqlite3.Connection: connection = sqlite3.connect( self.path, timeout=10, isolation_level=None, factory=ManagedConnection, ) connection.row_factory = sqlite3.Row connection.execute("PRAGMA foreign_keys = ON") connection.execute("PRAGMA busy_timeout = 5000") if self.path != ":memory:": connection.execute("PRAGMA journal_mode = WAL") connection.execute("PRAGMA synchronous = FULL") return connection def initialize(self) -> None: with self.connect() as connection: connection.executescript(SCHEMA) @contextmanager def transaction(self) -> Iterator[sqlite3.Connection]: connection = self.connect() try: connection.execute("BEGIN IMMEDIATE") yield connection connection.execute("COMMIT") except Exception: connection.execute("ROLLBACK") raise finally: connection.close() def row_to_dict(row: sqlite3.Row | None) -> dict | None: return dict(row) if row is not None else None ======================================================================================== FILE: app/schemas.py ======================================================================================== from __future__ import annotations from datetime import datetime from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field, field_validator class StrictModel(BaseModel): model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) class ResidentCreate(StrictModel): preferred_name: str = Field(min_length=1, max_length=100) locale: str = Field(default="en-CA", pattern=r"^[a-z]{2}-[A-Z]{2}$") timezone: str = Field(default="America/Toronto", min_length=3, max_length=80) quiet_start: str = Field(default="21:00", pattern=r"^(?:[01]\d|2[0-3]):[0-5]\d$") quiet_end: str = Field(default="07:00", pattern=r"^(?:[01]\d|2[0-3]):[0-5]\d$") class ContactCreate(StrictModel): display_name: str = Field(min_length=1, max_length=100) relationship: str = Field(min_length=1, max_length=100) destination_ref: str = Field( min_length=3, max_length=240, description="Opaque provider destination reference; do not send raw credentials.", ) allow_calls: bool = True allow_messages: bool = False call_window_start: str = Field(default="07:00", pattern=r"^(?:[01]\d|2[0-3]):[0-5]\d$") call_window_end: str = Field(default="21:00", pattern=r"^(?:[01]\d|2[0-3]):[0-5]\d$") disclosure: list[str] = Field(default_factory=list, max_length=20) class FactCreate(StrictModel): subject: str = Field(min_length=1, max_length=160) predicate: str = Field(min_length=1, max_length=120, pattern=r"^[a-z0-9_.-]+$") value: Any source_kind: Literal["caregiver", "resident", "calendar", "document", "system_import"] source_reference: str = Field(min_length=1, max_length=300) confidence: float = Field(default=1.0, ge=0, le=1) sensitivity: Literal["public", "personal", "sensitive", "restricted"] = "personal" allowed_purposes: list[str] = Field(default_factory=lambda: ["speak"], min_length=1, max_length=20) valid_from: datetime | None = None valid_until: datetime | None = None correction_of_id: str | None = None @field_validator("value") @classmethod def value_fits_fact_envelope(cls, value: Any) -> Any: import json try: encoded = json.dumps(value, ensure_ascii=False) except TypeError as error: raise ValueError("value must be JSON serializable") from error if len(encoded) > 4_000: raise ValueError("value exceeds the 4,000-character fact envelope") return value class FactVerify(StrictModel): verification_note: str | None = Field(default=None, max_length=500) class ReminderCreate(StrictModel): title: str = Field(min_length=1, max_length=240) due_at: datetime timezone: str = Field(default="America/Toronto", min_length=3, max_length=80) recurrence_rule: str | None = Field(default=None, max_length=500) source_reference: str = Field(min_length=1, max_length=300) @field_validator("due_at") @classmethod def due_at_must_be_timezone_aware(cls, value: datetime) -> datetime: if value.tzinfo is None: raise ValueError("due_at must include a timezone offset") return value class ReminderAcknowledge(StrictModel): note: str | None = Field(default=None, max_length=500) class DeviceCreate(StrictModel): display_name: str = Field(default="ALMA Presence Alpha", min_length=1, max_length=120) class DeviceEvent(StrictModel): event_type: Literal[ "privacy_shutter_closed", "microphone_disconnected", "motor_healthy", "network_online", ] value: bool class HomeEntityAllow(StrictModel): entity_id: str = Field(min_length=3, max_length=100, pattern=r"^[a-z_]+\.[a-z0-9_]+$") allowed_actions: list[Literal["home.light.set", "home.media.control"]] = Field( min_length=1, max_length=2, ) constraints: dict[str, Any] = Field(default_factory=dict) class ActionProposal(StrictModel): request_id: str = Field(min_length=8, max_length=120) resident_id: str = Field(min_length=8, max_length=80) device_id: str | None = Field(default=None, max_length=80) actor_kind: Literal["resident", "caregiver", "device", "assistant"] action_type: str = Field(min_length=3, max_length=100, pattern=r"^[a-z0-9_.-]+$") params: dict[str, Any] class ActionConfirm(StrictModel): token: str = Field(min_length=40, max_length=2_000) phrase: str = Field(min_length=1, max_length=30) ======================================================================================== FILE: app/policy.py ======================================================================================== from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta, timezone import base64 import hashlib import hmac import json from typing import Any ALLOWED_HOME_ACTIONS = {"home.light.set", "home.media.control"} COMMUNICATION_ACTIONS = {"call.place", "message.send"} COMMITMENT_ACTIONS = {"booking.create"} HOME_SECURITY_ACTIONS = {"door.unlock", "garage.open", "water.shutoff"} PROHIBITED_ACTIONS = { "finance.transfer", "purchase.create", "stove.activate", "oven.activate", "alarm.disable", "camera.remote.enable", "medication.dispense", } @dataclass(frozen=True) class PolicyContext: contact_allowed: bool = False home_entity_allowed: bool = False device_privacy_safe: bool = True @dataclass(frozen=True) class PolicyDecision: decision: str state: str risk_class: int reason_codes: tuple[str, ...] normalized_params: dict[str, Any] confirmation_prompt: str | None = None def utc_now() -> datetime: return datetime.now(timezone.utc) def isoformat(value: datetime) -> str: return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") def parse_utc(value: str) -> datetime: return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc) def canonical_json(value: Any) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def payload_hash( resident_id: str, device_id: str | None, action_type: str, normalized_params: dict[str, Any], ) -> str: payload = { "resident_id": resident_id, "device_id": device_id, "action_type": action_type, "params": normalized_params, } return hashlib.sha256(canonical_json(payload).encode("utf-8")).hexdigest() def token_hash(token: str) -> str: return hashlib.sha256(token.encode("utf-8")).hexdigest() class ConfirmationSigner: def __init__(self, secret: str, ttl_seconds: int = 120): if len(secret) < 24: raise ValueError("confirmation secret must contain at least 24 characters") self.secret = secret.encode("utf-8") self.ttl_seconds = ttl_seconds def issue(self, action_id: str, digest: str, now: datetime | None = None) -> tuple[str, str]: issued_at = now or utc_now() expires_at = issued_at + timedelta(seconds=self.ttl_seconds) claims = { "action_id": action_id, "payload_hash": digest, "expires_at": isoformat(expires_at), } body = base64.urlsafe_b64encode(canonical_json(claims).encode("utf-8")).decode("ascii").rstrip("=") signature = hmac.new(self.secret, body.encode("ascii"), hashlib.sha256).hexdigest() return f"{body}.{signature}", claims["expires_at"] def verify( self, token: str, action_id: str, digest: str, now: datetime | None = None, ) -> bool: try: body, supplied_signature = token.split(".", 1) expected_signature = hmac.new( self.secret, body.encode("ascii"), hashlib.sha256, ).hexdigest() if not hmac.compare_digest(expected_signature, supplied_signature): return False padded = body + "=" * (-len(body) % 4) claims = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8")) current = now or utc_now() return ( claims.get("action_id") == action_id and claims.get("payload_hash") == digest and current <= parse_utc(claims["expires_at"]) ) except (ValueError, KeyError, TypeError, json.JSONDecodeError): return False def _clean_text(value: Any, field: str, max_length: int) -> str: if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field} is required") cleaned = " ".join(value.strip().split()) if len(cleaned) > max_length: raise ValueError(f"{field} exceeds {max_length} characters") return cleaned class PolicyEngine: """Pure, deterministic action classifier. It never calls an external tool.""" def __init__(self, policy_version: str = "alpha-1"): self.policy_version = policy_version def evaluate( self, action_type: str, params: dict[str, Any], context: PolicyContext, ) -> PolicyDecision: action_type = action_type.strip().lower() try: normalized = self.normalize(action_type, params) except ValueError as error: return PolicyDecision( decision="blocked", state="blocked", risk_class=5, reason_codes=("invalid_parameters", str(error)), normalized_params={}, ) if action_type == "information.read": return PolicyDecision( decision="allowed", state="policy_checked", risk_class=0, reason_codes=("informational_only",), normalized_params=normalized, ) if action_type in ALLOWED_HOME_ACTIONS: if not context.home_entity_allowed: return PolicyDecision( decision="blocked", state="blocked", risk_class=5, reason_codes=("home_entity_not_allowlisted",), normalized_params=normalized, ) return PolicyDecision( decision="allowed", state="policy_checked", risk_class=1, reason_codes=("reversible_low_risk", "home_entity_allowlisted"), normalized_params=normalized, ) if action_type in COMMUNICATION_ACTIONS: if not context.contact_allowed: return PolicyDecision( decision="blocked", state="blocked", risk_class=5, reason_codes=("contact_not_trusted_or_not_permitted",), normalized_params=normalized, ) prompt = self._communication_prompt(action_type, normalized) return PolicyDecision( decision="confirmation_required", state="confirmation_required", risk_class=2, reason_codes=("external_communication", "trusted_contact"), normalized_params=normalized, confirmation_prompt=prompt, ) if action_type in COMMITMENT_ACTIONS: return PolicyDecision( decision="confirmation_required", state="confirmation_required", risk_class=3, reason_codes=("external_commitment", "exact_option_required"), normalized_params=normalized, confirmation_prompt=( f"Confirm booking with {normalized['provider_name']} at " f"{normalized['starts_at']}, price {normalized['price_summary']}, " f"with cancellation terms: {normalized['cancellation_terms']}." ), ) if action_type in HOME_SECURITY_ACTIONS: return PolicyDecision( decision="blocked", state="blocked", risk_class=4, reason_codes=("home_security_actuation_blocked_in_alpha",), normalized_params=normalized, ) if action_type in PROHIBITED_ACTIONS: return PolicyDecision( decision="blocked", state="blocked", risk_class=5, reason_codes=("prohibited_in_alpha",), normalized_params=normalized, ) return PolicyDecision( decision="blocked", state="blocked", risk_class=5, reason_codes=("unknown_action_type",), normalized_params=normalized, ) def normalize(self, action_type: str, params: dict[str, Any]) -> dict[str, Any]: if not isinstance(params, dict): raise ValueError("params must be an object") if action_type == "information.read": return {"resource": _clean_text(params.get("resource"), "resource", 100)} if action_type == "home.light.set": entity_id = _clean_text(params.get("entity_id"), "entity_id", 100).lower() state = _clean_text(params.get("state"), "state", 10).lower() if not entity_id.startswith("light.") or state not in {"on", "off"}: raise ValueError("light action requires light.* entity_id and on/off state") return {"entity_id": entity_id, "state": state} if action_type == "home.media.control": entity_id = _clean_text(params.get("entity_id"), "entity_id", 100).lower() command = _clean_text(params.get("command"), "command", 30).lower() if not entity_id.startswith("media_player.") or command not in { "play", "pause", "stop", "volume_up", "volume_down", }: raise ValueError("media action is outside the allowlisted command set") return {"entity_id": entity_id, "command": command} if action_type == "call.place": return {"contact_id": _clean_text(params.get("contact_id"), "contact_id", 80)} if action_type == "message.send": return { "contact_id": _clean_text(params.get("contact_id"), "contact_id", 80), "body": _clean_text(params.get("body"), "body", 500), } if action_type == "booking.create": return { "provider_name": _clean_text(params.get("provider_name"), "provider_name", 120), "starts_at": _clean_text(params.get("starts_at"), "starts_at", 50), "price_summary": _clean_text(params.get("price_summary"), "price_summary", 100), "cancellation_terms": _clean_text( params.get("cancellation_terms"), "cancellation_terms", 300, ), } # Blocked and unknown actions retain only compact, JSON-safe parameters for audit. encoded = canonical_json(params) if len(encoded) > 2_000: raise ValueError("params exceed the audit envelope") return json.loads(encoded) @staticmethod def _communication_prompt(action_type: str, params: dict[str, Any]) -> str: if action_type == "call.place": return f"Confirm placing a call to trusted contact {params['contact_id']}." return ( f"Confirm sending this exact message to trusted contact {params['contact_id']}: " f"{params['body']}" ) ======================================================================================== FILE: app/adapters.py ======================================================================================== from __future__ import annotations from dataclasses import asdict, dataclass from datetime import datetime, timezone import uuid from typing import Any, Protocol def now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") @dataclass(frozen=True) class ExecutionReceipt: action_type: str status: str provider: str provider_reference: str user_safe_summary: str completed_at: str reversible: bool def to_dict(self) -> dict[str, Any]: return asdict(self) class ActionAdapter(Protocol): def supports(self, action_type: str) -> bool: ... def execute(self, action_type: str, params: dict[str, Any]) -> ExecutionReceipt: ... class MockAlmaTalkAdapter: supported = {"call.place", "message.send", "booking.create"} def supports(self, action_type: str) -> bool: return action_type in self.supported def execute(self, action_type: str, params: dict[str, Any]) -> ExecutionReceipt: reference = f"mock-almatalk-{uuid.uuid4()}" if action_type == "call.place": summary = f"Demo call initiated to trusted contact {params['contact_id']}." reversible = False elif action_type == "message.send": summary = f"Demo message sent to trusted contact {params['contact_id']}." reversible = False elif action_type == "booking.create": summary = ( f"Demo booking created with {params['provider_name']} at " f"{params['starts_at']}." ) reversible = True else: raise ValueError("unsupported AlmaTalk action") return ExecutionReceipt( action_type=action_type, status="succeeded", provider="mock-almatalk", provider_reference=reference, user_safe_summary=summary, completed_at=now_iso(), reversible=reversible, ) class MockHomeAssistantAdapter: supported = {"home.light.set", "home.media.control"} def supports(self, action_type: str) -> bool: return action_type in self.supported def execute(self, action_type: str, params: dict[str, Any]) -> ExecutionReceipt: reference = f"mock-home-{uuid.uuid4()}" if action_type == "home.light.set": summary = f"{params['entity_id']} was turned {params['state']} in demo mode." elif action_type == "home.media.control": summary = f"{params['command']} was sent to {params['entity_id']} in demo mode." else: raise ValueError("unsupported Home Assistant action") return ExecutionReceipt( action_type=action_type, status="succeeded", provider="mock-home-assistant", provider_reference=reference, user_safe_summary=summary, completed_at=now_iso(), reversible=True, ) class LocalInformationAdapter: supported = {"information.read"} def supports(self, action_type: str) -> bool: return action_type in self.supported def execute(self, action_type: str, params: dict[str, Any]) -> ExecutionReceipt: if action_type != "information.read": raise ValueError("unsupported local information action") return ExecutionReceipt( action_type=action_type, status="succeeded", provider="local-information", provider_reference=f"local-{uuid.uuid4()}", user_safe_summary=f"Local information resource {params['resource']} was read.", completed_at=now_iso(), reversible=True, ) class ToolRegistry: """Closed adapter registry; it intentionally has no generic HTTP or shell adapter.""" def __init__(self) -> None: self.adapters: tuple[ActionAdapter, ...] = ( LocalInformationAdapter(), MockHomeAssistantAdapter(), MockAlmaTalkAdapter(), ) def execute(self, action_type: str, params: dict[str, Any]) -> ExecutionReceipt: for adapter in self.adapters: if adapter.supports(action_type): return adapter.execute(action_type, params) raise ValueError(f"no reviewed adapter registered for {action_type}") ======================================================================================== FILE: app/services.py ======================================================================================== from __future__ import annotations from datetime import datetime, timezone import json import sqlite3 from typing import Any import uuid from .adapters import ToolRegistry from .database import Database, row_to_dict from .policy import ( ConfirmationSigner, PolicyContext, PolicyEngine, canonical_json, parse_utc, payload_hash, token_hash, utc_now, ) def now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def require_row(row: sqlite3.Row | None, label: str) -> sqlite3.Row: if row is None: raise KeyError(f"{label} not found") return row def append_audit( connection: sqlite3.Connection, *, resident_id: str | None, actor_kind: str, event_type: str, entity_type: str, entity_id: str, details: dict[str, Any], ) -> str: audit_id = str(uuid.uuid4()) connection.execute( """ INSERT INTO audit_events (id, resident_id, actor_kind, event_type, entity_type, entity_id, details_json, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( audit_id, resident_id, actor_kind, event_type, entity_type, entity_id, canonical_json(details), now_iso(), ), ) return audit_id class ResidentService: def __init__(self, database: Database): self.database = database def create( self, *, preferred_name: str, locale: str, timezone_name: str, quiet_start: str, quiet_end: str, actor_kind: str, ) -> dict[str, Any]: resident_id = str(uuid.uuid4()) created_at = now_iso() with self.database.transaction() as connection: connection.execute( """ INSERT INTO residents (id, preferred_name, locale, timezone, quiet_start, quiet_end, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( resident_id, preferred_name, locale, timezone_name, quiet_start, quiet_end, created_at, ), ) append_audit( connection, resident_id=resident_id, actor_kind=actor_kind, event_type="resident.created", entity_type="resident", entity_id=resident_id, details={"locale": locale, "timezone": timezone_name}, ) return { "id": resident_id, "preferred_name": preferred_name, "locale": locale, "timezone": timezone_name, "quiet_start": quiet_start, "quiet_end": quiet_end, "created_at": created_at, } def get(self, resident_id: str) -> dict[str, Any]: with self.database.connect() as connection: row = require_row( connection.execute( "SELECT * FROM residents WHERE id = ?", (resident_id,), ).fetchone(), "resident", ) return dict(row) class ContactService: def __init__(self, database: Database): self.database = database def create( self, *, resident_id: str, display_name: str, relationship: str, destination_ref: str, allow_calls: bool, allow_messages: bool, call_window_start: str, call_window_end: str, disclosure: list[str], actor_kind: str, ) -> dict[str, Any]: contact_id = str(uuid.uuid4()) created_at = now_iso() with self.database.transaction() as connection: require_row( connection.execute("SELECT id FROM residents WHERE id = ?", (resident_id,)).fetchone(), "resident", ) connection.execute( """ INSERT INTO trusted_contacts (id, resident_id, display_name, relationship, destination_ref, call_window_start, call_window_end, allow_calls, allow_messages, disclosure_json, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( contact_id, resident_id, display_name, relationship, destination_ref, call_window_start, call_window_end, int(allow_calls), int(allow_messages), canonical_json(disclosure), created_at, ), ) append_audit( connection, resident_id=resident_id, actor_kind=actor_kind, event_type="trusted_contact.created", entity_type="trusted_contact", entity_id=contact_id, details={ "display_name": display_name, "relationship": relationship, "allow_calls": allow_calls, "allow_messages": allow_messages, }, ) return { "id": contact_id, "resident_id": resident_id, "display_name": display_name, "relationship": relationship, "allow_calls": allow_calls, "allow_messages": allow_messages, "call_window_start": call_window_start, "call_window_end": call_window_end, "disclosure": disclosure, "created_at": created_at, } class HomeEntityService: def __init__(self, database: Database): self.database = database def allow( self, *, resident_id: str, entity_id: str, allowed_actions: list[str], constraints: dict[str, Any], actor_kind: str, ) -> dict[str, Any]: row_id = str(uuid.uuid4()) created_at = now_iso() with self.database.transaction() as connection: require_row( connection.execute("SELECT id FROM residents WHERE id = ?", (resident_id,)).fetchone(), "resident", ) connection.execute( """ INSERT INTO home_entity_allowlist (id, resident_id, entity_id, allowed_actions_json, value_constraints_json, created_at) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(resident_id, entity_id) DO UPDATE SET allowed_actions_json = excluded.allowed_actions_json, value_constraints_json = excluded.value_constraints_json """, ( row_id, resident_id, entity_id, canonical_json(sorted(set(allowed_actions))), canonical_json(constraints), created_at, ), ) append_audit( connection, resident_id=resident_id, actor_kind=actor_kind, event_type="home_entity.allowlisted", entity_type="home_entity", entity_id=entity_id, details={"allowed_actions": sorted(set(allowed_actions)), "constraints": constraints}, ) return { "resident_id": resident_id, "entity_id": entity_id, "allowed_actions": sorted(set(allowed_actions)), "constraints": constraints, } class MemoryService: def __init__(self, database: Database): self.database = database def create_fact( self, *, resident_id: str, subject: str, predicate: str, value: Any, source_kind: str, source_reference: str, created_by: str, confidence: float, sensitivity: str, allowed_purposes: list[str], valid_from: str | None, valid_until: str | None, correction_of_id: str | None = None, ) -> dict[str, Any]: fact_id = str(uuid.uuid4()) timestamp = now_iso() with self.database.transaction() as connection: require_row( connection.execute("SELECT id FROM residents WHERE id = ?", (resident_id,)).fetchone(), "resident", ) if correction_of_id: require_row( connection.execute( "SELECT id FROM memory_facts WHERE id = ? AND resident_id = ?", (correction_of_id, resident_id), ).fetchone(), "fact being corrected", ) connection.execute( """ INSERT INTO memory_facts (id, resident_id, subject, predicate, value_json, source_kind, source_reference, created_by, verification_status, confidence, sensitivity, allowed_purposes_json, valid_from, valid_until, correction_of_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'proposed', ?, ?, ?, ?, ?, ?, ?, ?) """, ( fact_id, resident_id, subject, predicate, canonical_json(value), source_kind, source_reference, created_by, confidence, sensitivity, canonical_json(sorted(set(allowed_purposes))), valid_from, valid_until, correction_of_id, timestamp, timestamp, ), ) append_audit( connection, resident_id=resident_id, actor_kind=created_by, event_type="memory_fact.proposed", entity_type="memory_fact", entity_id=fact_id, details={ "predicate": predicate, "source_kind": source_kind, "source_reference": source_reference, "correction_of_id": correction_of_id, }, ) return self.get_fact(fact_id) def verify_fact(self, fact_id: str, verified_by: str) -> dict[str, Any]: timestamp = now_iso() with self.database.transaction() as connection: row = require_row( connection.execute("SELECT * FROM memory_facts WHERE id = ?", (fact_id,)).fetchone(), "memory fact", ) if row["verification_status"] not in {"proposed", "verified"}: raise ValueError("only proposed or already verified facts can be verified") connection.execute( """ UPDATE memory_facts SET verification_status = 'verified', verified_by = ?, verified_at = ?, updated_at = ? WHERE id = ? """, (verified_by, timestamp, timestamp, fact_id), ) if row["correction_of_id"]: connection.execute( """ UPDATE memory_facts SET verification_status = 'rejected', updated_at = ? WHERE id = ? """, (timestamp, row["correction_of_id"]), ) append_audit( connection, resident_id=row["resident_id"], actor_kind=verified_by, event_type="memory_fact.verified", entity_type="memory_fact", entity_id=fact_id, details={"correction_of_id": row["correction_of_id"]}, ) return self.get_fact(fact_id) def get_fact(self, fact_id: str) -> dict[str, Any]: with self.database.connect() as connection: row = require_row( connection.execute("SELECT * FROM memory_facts WHERE id = ?", (fact_id,)).fetchone(), "memory fact", ) return self._serialize_fact(row) def query_verified( self, *, resident_id: str, predicate: str | None, purpose: str, at: datetime | None = None, ) -> list[dict[str, Any]]: current = at or utc_now() query = "SELECT * FROM memory_facts WHERE resident_id = ? AND verification_status = 'verified'" params: list[Any] = [resident_id] if predicate: query += " AND predicate = ?" params.append(predicate) query += " ORDER BY verified_at DESC, created_at DESC LIMIT 50" with self.database.connect() as connection: rows = connection.execute(query, params).fetchall() results = [] for row in rows: allowed_purposes = json.loads(row["allowed_purposes_json"]) if purpose not in allowed_purposes: continue if row["valid_from"] and current < parse_utc(row["valid_from"]): continue if row["valid_until"] and current > parse_utc(row["valid_until"]): continue results.append(self._serialize_fact(row)) return results @staticmethod def _serialize_fact(row: sqlite3.Row) -> dict[str, Any]: result = dict(row) result["value"] = json.loads(result.pop("value_json")) result["allowed_purposes"] = json.loads(result.pop("allowed_purposes_json")) return result class ReminderService: def __init__(self, database: Database): self.database = database def create( self, *, resident_id: str, title: str, due_at: str, timezone_name: str, recurrence_rule: str | None, source_reference: str, actor_kind: str, ) -> dict[str, Any]: reminder_id = str(uuid.uuid4()) timestamp = now_iso() with self.database.transaction() as connection: require_row( connection.execute("SELECT id FROM residents WHERE id = ?", (resident_id,)).fetchone(), "resident", ) connection.execute( """ INSERT INTO reminders (id, resident_id, title, due_at, timezone, recurrence_rule, source_reference, status, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, 'scheduled', ?, ?) """, ( reminder_id, resident_id, title, due_at, timezone_name, recurrence_rule, source_reference, timestamp, timestamp, ), ) append_audit( connection, resident_id=resident_id, actor_kind=actor_kind, event_type="reminder.created", entity_type="reminder", entity_id=reminder_id, details={"title": title, "due_at": due_at, "source_reference": source_reference}, ) return self.get(reminder_id) def due(self, resident_id: str, at: str | None = None) -> list[dict[str, Any]]: current = at or now_iso() with self.database.connect() as connection: rows = connection.execute( """ SELECT * FROM reminders WHERE resident_id = ? AND status IN ('scheduled', 'snoozed') AND due_at <= ? ORDER BY due_at ASC """, (resident_id, current), ).fetchall() return [dict(row) for row in rows] def acknowledge(self, reminder_id: str, note: str | None, actor_kind: str) -> dict[str, Any]: timestamp = now_iso() with self.database.transaction() as connection: row = require_row( connection.execute("SELECT * FROM reminders WHERE id = ?", (reminder_id,)).fetchone(), "reminder", ) connection.execute( """ UPDATE reminders SET status = 'acknowledged', acknowledgement_note = ?, updated_at = ? WHERE id = ? """, (note, timestamp, reminder_id), ) append_audit( connection, resident_id=row["resident_id"], actor_kind=actor_kind, event_type="reminder.acknowledged", entity_type="reminder", entity_id=reminder_id, details={ "acknowledged_at": timestamp, "note": note, "interpretation_boundary": "acknowledgement is not evidence of medication adherence", }, ) return self.get(reminder_id) def get(self, reminder_id: str) -> dict[str, Any]: with self.database.connect() as connection: return dict( require_row( connection.execute("SELECT * FROM reminders WHERE id = ?", (reminder_id,)).fetchone(), "reminder", ) ) class DeviceService: ALLOWED_EVENTS = { "privacy_shutter_closed": "privacy_shutter_closed", "microphone_disconnected": "microphone_disconnected", "motor_healthy": "motor_healthy", "network_online": "network_online", } def __init__(self, database: Database): self.database = database def register(self, resident_id: str, display_name: str, actor_kind: str) -> dict[str, Any]: device_id = str(uuid.uuid4()) created_at = now_iso() with self.database.transaction() as connection: require_row( connection.execute("SELECT id FROM residents WHERE id = ?", (resident_id,)).fetchone(), "resident", ) connection.execute( "INSERT INTO devices (id, resident_id, display_name, created_at) VALUES (?, ?, ?, ?)", (device_id, resident_id, display_name, created_at), ) append_audit( connection, resident_id=resident_id, actor_kind=actor_kind, event_type="device.registered", entity_type="device", entity_id=device_id, details={"display_name": display_name}, ) return self.get(device_id) def record_event( self, device_id: str, event_type: str, value: bool, actor_kind: str = "hardware", ) -> dict[str, Any]: column = self.ALLOWED_EVENTS.get(event_type) if not column: raise ValueError("unsupported device event") timestamp = now_iso() with self.database.transaction() as connection: row = require_row( connection.execute("SELECT * FROM devices WHERE id = ?", (device_id,)).fetchone(), "device", ) # Column is selected from a closed constant map, never user input. connection.execute( f"UPDATE devices SET {column} = ?, last_seen_at = ? WHERE id = ?", # noqa: S608 (int(value), timestamp, device_id), ) append_audit( connection, resident_id=row["resident_id"], actor_kind=actor_kind, event_type=f"device.{event_type}", entity_type="device", entity_id=device_id, details={"value": value, "observed_at": timestamp}, ) return self.get(device_id) def get(self, device_id: str) -> dict[str, Any]: with self.database.connect() as connection: return dict( require_row( connection.execute("SELECT * FROM devices WHERE id = ?", (device_id,)).fetchone(), "device", ) ) class ActionService: def __init__( self, database: Database, policy: PolicyEngine, signer: ConfirmationSigner, tools: ToolRegistry, ) -> None: self.database = database self.policy = policy self.signer = signer self.tools = tools def propose( self, *, request_id: str, resident_id: str, device_id: str | None, actor_kind: str, action_type: str, params: dict[str, Any], ) -> dict[str, Any]: with self.database.connect() as connection: existing = connection.execute( "SELECT * FROM action_proposals WHERE request_id = ?", (request_id,), ).fetchone() if existing: result = self._serialize_action(existing) result["idempotent_replay"] = True return result context = self._policy_context(resident_id, device_id, action_type, params) decision = self.policy.evaluate(action_type, params, context) action_id = str(uuid.uuid4()) digest = payload_hash(resident_id, device_id, action_type, decision.normalized_params) timestamp = now_iso() confirmation_token = None confirmation_token_digest = None confirmation_expires_at = None if decision.decision == "confirmation_required": confirmation_token, confirmation_expires_at = self.signer.issue(action_id, digest) confirmation_token_digest = token_hash(confirmation_token) with self.database.transaction() as connection: require_row( connection.execute("SELECT id FROM residents WHERE id = ?", (resident_id,)).fetchone(), "resident", ) if device_id: require_row( connection.execute( "SELECT id FROM devices WHERE id = ? AND resident_id = ?", (device_id, resident_id), ).fetchone(), "device", ) connection.execute( """ INSERT INTO action_proposals (id, request_id, resident_id, device_id, actor_kind, action_type, normalized_params_json, payload_hash, risk_class, state, decision, reason_codes_json, policy_version, confirmation_token_hash, confirmation_expires_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( action_id, request_id, resident_id, device_id, actor_kind, action_type, canonical_json(decision.normalized_params), digest, decision.risk_class, decision.state, decision.decision, canonical_json(list(decision.reason_codes)), self.policy.policy_version, confirmation_token_digest, confirmation_expires_at, timestamp, timestamp, ), ) append_audit( connection, resident_id=resident_id, actor_kind=actor_kind, event_type="action.policy_decided", entity_type="action", entity_id=action_id, details={ "action_type": action_type, "decision": decision.decision, "risk_class": decision.risk_class, "reason_codes": list(decision.reason_codes), "payload_hash": digest, "policy_version": self.policy.policy_version, }, ) if decision.decision == "allowed": result = self._execute(action_id, confirmed=False) else: result = self.get(action_id) if confirmation_token: result["confirmation"] = { "token": confirmation_token, "expires_at": confirmation_expires_at, "prompt": decision.confirmation_prompt, "payload_hash": digest, } return result def confirm(self, action_id: str, token: str, phrase: str, actor_kind: str) -> dict[str, Any]: if phrase.strip().lower() != "confirm": raise ValueError("exact confirmation phrase is required") with self.database.connect() as connection: row = require_row( connection.execute("SELECT * FROM action_proposals WHERE id = ?", (action_id,)).fetchone(), "action", ) if row["state"] in {"succeeded", "failed"}: result = self._serialize_action(row) result["idempotent_replay"] = True return result if row["state"] != "confirmation_required": raise ValueError("action is not awaiting confirmation") if row["confirmation_token_hash"] != token_hash(token): raise PermissionError("confirmation token does not match the stored action") if not self.signer.verify(token, action_id, row["payload_hash"]): self._expire(action_id, row["resident_id"], actor_kind) raise PermissionError("confirmation token is invalid or expired") with self.database.transaction() as connection: connection.execute( """ UPDATE action_proposals SET state = 'confirmed', confirmed_at = ?, updated_at = ? WHERE id = ? AND state = 'confirmation_required' """, (now_iso(), now_iso(), action_id), ) append_audit( connection, resident_id=row["resident_id"], actor_kind=actor_kind, event_type="action.confirmed", entity_type="action", entity_id=action_id, details={"payload_hash": row["payload_hash"]}, ) return self._execute(action_id, confirmed=True) def get(self, action_id: str) -> dict[str, Any]: with self.database.connect() as connection: row = require_row( connection.execute("SELECT * FROM action_proposals WHERE id = ?", (action_id,)).fetchone(), "action", ) return self._serialize_action(row) def receipt(self, action_id: str) -> dict[str, Any]: action = self.get(action_id) if not action.get("receipt"): raise ValueError("action has no execution receipt") return { "action_id": action_id, "state": action["state"], "payload_hash": action["payload_hash"], "receipt": action["receipt"], } def _execute(self, action_id: str, confirmed: bool) -> dict[str, Any]: with self.database.transaction() as connection: row = require_row( connection.execute("SELECT * FROM action_proposals WHERE id = ?", (action_id,)).fetchone(), "action", ) if row["state"] == "succeeded": return self._serialize_action(row) allowed_states = {"policy_checked"} if not confirmed else {"confirmed"} if row["state"] not in allowed_states: raise ValueError("action state does not permit execution") connection.execute( "UPDATE action_proposals SET state = 'executing', updated_at = ? WHERE id = ?", (now_iso(), action_id), ) try: receipt = self.tools.execute( row["action_type"], json.loads(row["normalized_params_json"]), ).to_dict() state = "succeeded" except Exception as error: receipt = { "status": "failed", "user_safe_summary": "The requested action could not be completed.", "error_type": type(error).__name__, "completed_at": now_iso(), } state = "failed" with self.database.transaction() as connection: connection.execute( """ UPDATE action_proposals SET state = ?, receipt_json = ?, updated_at = ? WHERE id = ? """, (state, canonical_json(receipt), now_iso(), action_id), ) append_audit( connection, resident_id=row["resident_id"], actor_kind="tool_gateway", event_type=f"action.{state}", entity_type="action", entity_id=action_id, details={ "action_type": row["action_type"], "payload_hash": row["payload_hash"], "receipt": receipt, }, ) return self.get(action_id) def _expire(self, action_id: str, resident_id: str, actor_kind: str) -> None: with self.database.transaction() as connection: connection.execute( "UPDATE action_proposals SET state = 'expired', updated_at = ? WHERE id = ?", (now_iso(), action_id), ) append_audit( connection, resident_id=resident_id, actor_kind=actor_kind, event_type="action.expired", entity_type="action", entity_id=action_id, details={}, ) def _policy_context( self, resident_id: str, device_id: str | None, action_type: str, params: dict[str, Any], ) -> PolicyContext: contact_allowed = False home_entity_allowed = False device_privacy_safe = True with self.database.connect() as connection: contact_id = params.get("contact_id") if isinstance(params, dict) else None if isinstance(contact_id, str): contact = connection.execute( """ SELECT allow_calls, allow_messages FROM trusted_contacts WHERE id = ? AND resident_id = ? """, (contact_id, resident_id), ).fetchone() if contact: if action_type == "call.place": contact_allowed = bool(contact["allow_calls"]) elif action_type == "message.send": contact_allowed = bool(contact["allow_messages"]) entity_id = params.get("entity_id") if isinstance(params, dict) else None if isinstance(entity_id, str): entity = connection.execute( """ SELECT allowed_actions_json FROM home_entity_allowlist WHERE resident_id = ? AND entity_id = ? """, (resident_id, entity_id.lower()), ).fetchone() if entity: home_entity_allowed = action_type in json.loads(entity["allowed_actions_json"]) if device_id: device = connection.execute( """ SELECT privacy_shutter_closed, microphone_disconnected FROM devices WHERE id = ? AND resident_id = ? """, (device_id, resident_id), ).fetchone() if device: device_privacy_safe = not ( action_type.startswith("vision.") and device["privacy_shutter_closed"] ) return PolicyContext( contact_allowed=contact_allowed, home_entity_allowed=home_entity_allowed, device_privacy_safe=device_privacy_safe, ) @staticmethod def _serialize_action(row: sqlite3.Row) -> dict[str, Any]: result = dict(row) result["params"] = json.loads(result.pop("normalized_params_json")) result["reason_codes"] = json.loads(result.pop("reason_codes_json")) receipt_json = result.pop("receipt_json") result["receipt"] = json.loads(receipt_json) if receipt_json else None result.pop("confirmation_token_hash", None) return result class AuditService: def __init__(self, database: Database): self.database = database def list_for_resident(self, resident_id: str, limit: int = 100) -> list[dict[str, Any]]: with self.database.connect() as connection: rows = connection.execute( """ SELECT * FROM audit_events WHERE resident_id = ? ORDER BY created_at DESC LIMIT ? """, (resident_id, min(max(limit, 1), 500)), ).fetchall() result = [] for row in rows: item = dict(row) item["details"] = json.loads(item.pop("details_json")) result.append(item) return result ======================================================================================== FILE: app/main.py ======================================================================================== from __future__ import annotations from contextlib import asynccontextmanager from typing import Annotated from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request from fastapi.responses import JSONResponse from .adapters import ToolRegistry from .database import Database from .policy import ConfirmationSigner, PolicyEngine from .schemas import ( ActionConfirm, ActionProposal, ContactCreate, DeviceCreate, DeviceEvent, FactCreate, FactVerify, HomeEntityAllow, ReminderAcknowledge, ReminderCreate, ResidentCreate, ) from .services import ( ActionService, AuditService, ContactService, DeviceService, HomeEntityService, MemoryService, ReminderService, ResidentService, ) from .settings import load_settings settings = load_settings() database = Database(settings.database_path) residents = ResidentService(database) contacts = ContactService(database) home_entities = HomeEntityService(database) memories = MemoryService(database) reminders = ReminderService(database) devices = DeviceService(database) actions = ActionService( database=database, policy=PolicyEngine(settings.policy_version), signer=ConfirmationSigner( settings.confirmation_secret, settings.confirmation_ttl_seconds, ), tools=ToolRegistry(), ) audit = AuditService(database) @asynccontextmanager async def lifespan(_: FastAPI): database.initialize() yield app = FastAPI( title="ALMA Presence Functional Alpha API", version="0.1.0", description=( "Local-first reference backend for verified memory, offline reminders, " "device privacy state and deterministic action safety. All external " "adapters run in mock mode in this package." ), lifespan=lifespan, docs_url="/docs", redoc_url="/redoc", ) def require_api_key( x_alma_api_key: Annotated[str | None, Header()] = None, ) -> None: if not x_alma_api_key or x_alma_api_key != settings.api_key: raise HTTPException(status_code=401, detail="valid X-ALMA-API-Key is required") def actor_from_header( x_alma_actor: Annotated[str | None, Header()] = None, ) -> str: actor = (x_alma_actor or "caregiver").strip().lower() if actor not in {"resident", "caregiver", "device", "hardware", "support"}: raise HTTPException(status_code=400, detail="unsupported X-ALMA-Actor") return actor protected = [Depends(require_api_key)] @app.exception_handler(KeyError) async def not_found_handler(_: Request, error: KeyError) -> JSONResponse: return JSONResponse(status_code=404, content={"error": str(error).strip("'")}) @app.exception_handler(PermissionError) async def permission_handler(_: Request, error: PermissionError) -> JSONResponse: return JSONResponse(status_code=403, content={"error": str(error)}) @app.exception_handler(ValueError) async def value_handler(_: Request, error: ValueError) -> JSONResponse: return JSONResponse(status_code=400, content={"error": str(error)}) @app.get("/health") def health() -> dict: return { "status": "ok", "service": "alma-presence-functional-alpha", "execution_mode": settings.execution_mode, "policy_version": settings.policy_version, "safety_notice": "real-world adapters are disabled; mock execution only", } @app.get("/") def root() -> dict: return { "name": "ALMA Presence Functional Alpha API", "docs": "/docs", "health": "/health", "principle": "the language model proposes; deterministic software decides", } @app.post("/v1/residents", status_code=201, dependencies=protected) def create_resident( payload: ResidentCreate, actor: Annotated[str, Depends(actor_from_header)], ) -> dict: return residents.create( preferred_name=payload.preferred_name, locale=payload.locale, timezone_name=payload.timezone, quiet_start=payload.quiet_start, quiet_end=payload.quiet_end, actor_kind=actor, ) @app.get("/v1/residents/{resident_id}", dependencies=protected) def get_resident(resident_id: str) -> dict: return residents.get(resident_id) @app.post("/v1/residents/{resident_id}/contacts", status_code=201, dependencies=protected) def create_contact( resident_id: str, payload: ContactCreate, actor: Annotated[str, Depends(actor_from_header)], ) -> dict: return contacts.create( resident_id=resident_id, display_name=payload.display_name, relationship=payload.relationship, destination_ref=payload.destination_ref, allow_calls=payload.allow_calls, allow_messages=payload.allow_messages, call_window_start=payload.call_window_start, call_window_end=payload.call_window_end, disclosure=payload.disclosure, actor_kind=actor, ) @app.put("/v1/residents/{resident_id}/home-entities/{entity_id}", dependencies=protected) def allow_home_entity( resident_id: str, entity_id: str, payload: HomeEntityAllow, actor: Annotated[str, Depends(actor_from_header)], ) -> dict: if entity_id != payload.entity_id: raise ValueError("path entity_id must match payload entity_id") return home_entities.allow( resident_id=resident_id, entity_id=payload.entity_id, allowed_actions=payload.allowed_actions, constraints=payload.constraints, actor_kind=actor, ) @app.post("/v1/residents/{resident_id}/facts", status_code=201, dependencies=protected) def create_fact( resident_id: str, payload: FactCreate, actor: Annotated[str, Depends(actor_from_header)], ) -> dict: return memories.create_fact( resident_id=resident_id, subject=payload.subject, predicate=payload.predicate, value=payload.value, source_kind=payload.source_kind, source_reference=payload.source_reference, created_by=actor, confidence=payload.confidence, sensitivity=payload.sensitivity, allowed_purposes=payload.allowed_purposes, valid_from=payload.valid_from.isoformat().replace("+00:00", "Z") if payload.valid_from else None, valid_until=payload.valid_until.isoformat().replace("+00:00", "Z") if payload.valid_until else None, correction_of_id=payload.correction_of_id, ) @app.post("/v1/facts/{fact_id}/verify", dependencies=protected) def verify_fact( fact_id: str, _: FactVerify, actor: Annotated[str, Depends(actor_from_header)], ) -> dict: if actor not in {"caregiver", "resident"}: raise HTTPException(status_code=403, detail="resident or caregiver verification is required") return memories.verify_fact(fact_id, actor) @app.get("/v1/residents/{resident_id}/facts/query", dependencies=protected) def query_facts( resident_id: str, predicate: str | None = None, purpose: str = Query(default="speak", min_length=1, max_length=80), ) -> dict: facts = memories.query_verified( resident_id=resident_id, predicate=predicate, purpose=purpose, ) return {"facts": facts, "count": len(facts), "verification_required": True} @app.post("/v1/residents/{resident_id}/reminders", status_code=201, dependencies=protected) def create_reminder( resident_id: str, payload: ReminderCreate, actor: Annotated[str, Depends(actor_from_header)], ) -> dict: return reminders.create( resident_id=resident_id, title=payload.title, due_at=payload.due_at.isoformat(), timezone_name=payload.timezone, recurrence_rule=payload.recurrence_rule, source_reference=payload.source_reference, actor_kind=actor, ) @app.get("/v1/residents/{resident_id}/reminders/due", dependencies=protected) def due_reminders(resident_id: str, at: str | None = None) -> dict: due = reminders.due(resident_id, at) return {"reminders": due, "count": len(due)} @app.post("/v1/reminders/{reminder_id}/acknowledge", dependencies=protected) def acknowledge_reminder( reminder_id: str, payload: ReminderAcknowledge, actor: Annotated[str, Depends(actor_from_header)], ) -> dict: result = reminders.acknowledge(reminder_id, payload.note, actor) result["interpretation_boundary"] = ( "This records acknowledgement only. It does not claim medication was taken." ) return result @app.post("/v1/residents/{resident_id}/devices", status_code=201, dependencies=protected) def register_device( resident_id: str, payload: DeviceCreate, actor: Annotated[str, Depends(actor_from_header)], ) -> dict: return devices.register(resident_id, payload.display_name, actor) @app.post("/v1/devices/{device_id}/events", dependencies=protected) def record_device_event( device_id: str, payload: DeviceEvent, actor: Annotated[str, Depends(actor_from_header)], ) -> dict: return devices.record_event(device_id, payload.event_type, payload.value, actor) @app.get("/v1/devices/{device_id}/health", dependencies=protected) def device_health(device_id: str) -> dict: device = devices.get(device_id) return { "device_id": device_id, "privacy_shutter_closed": bool(device["privacy_shutter_closed"]), "microphone_disconnected": bool(device["microphone_disconnected"]), "motor_healthy": bool(device["motor_healthy"]), "network_online": bool(device["network_online"]), "last_seen_at": device["last_seen_at"], } @app.post("/v1/actions/proposals", status_code=201, dependencies=protected) def propose_action(payload: ActionProposal) -> dict: return actions.propose( request_id=payload.request_id, resident_id=payload.resident_id, device_id=payload.device_id, actor_kind=payload.actor_kind, action_type=payload.action_type, params=payload.params, ) @app.post("/v1/actions/{action_id}/confirm", dependencies=protected) def confirm_action( action_id: str, payload: ActionConfirm, actor: Annotated[str, Depends(actor_from_header)], ) -> dict: return actions.confirm(action_id, payload.token, payload.phrase, actor) @app.get("/v1/actions/{action_id}", dependencies=protected) def get_action(action_id: str) -> dict: return actions.get(action_id) @app.get("/v1/actions/{action_id}/receipt", dependencies=protected) def get_receipt(action_id: str) -> dict: return actions.receipt(action_id) @app.get("/v1/residents/{resident_id}/audit", dependencies=protected) def list_audit( resident_id: str, limit: int = Query(default=100, ge=1, le=500), ) -> dict: events = audit.list_for_resident(resident_id, limit) return {"events": events, "count": len(events), "append_only": True} @app.get("/v1/avatar/contract", dependencies=protected) def avatar_contract() -> dict: return { "version": "1.0", "states": [ "offline", "sleeping", "available", "greeting", "listening", "thinking", "speaking", "calling", "reassuring", "celebrating", "error", "privacy_mode", ], "required_inputs": [ "speech_level", "mouth_pose", "eye_horizontal", "eye_vertical", "blink", "expression", "expression_intensity", "gesture", "status", "gaze_target", ], "fallbacks": ["reduced_motion", "static_portrait", "captions", "no_avatar"], } ======================================================================================== FILE: contracts/action-proposal.schema.json ======================================================================================== { "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "https://almaintelligence.ai/contracts/alma-presence/action-proposal-v1.json", "title": "ALMA Presence Action Proposal v1", "type": "object", "additionalProperties": false, "required": ["request_id", "resident_id", "actor_kind", "action_type", "params"], "properties": { "request_id": { "type": "string", "minLength": 8, "maxLength": 120 }, "resident_id": { "type": "string", "minLength": 8, "maxLength": 80 }, "device_id": { "type": ["string", "null"], "maxLength": 80 }, "actor_kind": { "enum": ["resident", "caregiver", "device", "assistant"] }, "action_type": { "type": "string", "pattern": "^[a-z0-9_.-]+$", "maxLength": 100 }, "params": { "type": "object" } } } ======================================================================================== FILE: contracts/avatar-contract-v1.json ======================================================================================== { "id": "alma-avatar-contract", "version": "1.0", "states": [ "offline", "sleeping", "available", "greeting", "listening", "thinking", "speaking", "calling", "reassuring", "celebrating", "error", "privacy_mode" ], "inputs": { "speech_level": { "type": "number", "minimum": 0, "maximum": 1 }, "mouth_pose": { "type": "string" }, "eye_horizontal": { "type": "number", "minimum": -1, "maximum": 1 }, "eye_vertical": { "type": "number", "minimum": -1, "maximum": 1 }, "blink": { "type": "boolean" }, "expression": { "type": "string" }, "expression_intensity": { "type": "number", "minimum": 0, "maximum": 0.6 }, "gesture": { "type": "string" }, "status": { "type": "string" }, "gaze_target": { "type": "string" } }, "fallbacks": ["reduced_motion", "static_portrait", "captions", "no_avatar"], "identity_boundary": "The avatar always identifies as an AI assistant and never impersonates a caller, clinician or family member." } ======================================================================================== FILE: hardware/README.md ======================================================================================== # Edge and safety-controller contract The Linux edge computer and the motor controller are separate trust zones. The motor controller must default to a safe static state if Linux, the AI runtime, serial communication, encoder feedback or the host heartbeat is lost. ## Host-to-controller messages Use a framed, checksummed protocol rather than raw servo commands. A minimal prototype envelope is: ```json { "version": 1, "sequence": 1042, "command": "pan_to", "target_degrees": 24, "max_speed_dps": 12, "max_current_ma": 500, "expires_ms": 500, "crc32": "..." } ``` Permitted commands are `heartbeat`, `pan_to`, `return_neutral`, `stop` and `read_status`. There is no generic register-write or firmware-shell command in the application protocol. ## Independent controller duties - Enforce physical and configured angular limits. - Check both mechanical end switches. - Limit speed, acceleration and current. - Stop on obstruction, encoder disagreement, overtemperature or undervoltage. - Stop if a valid heartbeat is absent for 500 ms. - Keep the motor branch off after an emergency-stop event until a physical reset and explicit reinitialization. - Report `static_safe` when movement is unavailable; the screen and voice experience must continue without motion. ## Privacy GPIO The camera-shutter and microphone-disconnect states must be independently sensed by the controller or a dedicated input device. The UI may report only the measured hardware state. A software mute icon is not proof that microphone power has been removed. ## Prototype boundary This contract is an engineering starting point, not a certified safety system. Validate pinch force, torque limits, anti-tip stability, thermal behaviour, electrical protection, electromagnetic compatibility and failure recovery with qualified specialists before any home pilot. ======================================================================================== FILE: scripts/demo.py ======================================================================================== #!/usr/bin/env python3 """Run the ALMA Presence magic-moments API demo against a local server.""" from __future__ import annotations import json import os import urllib.error import urllib.request import uuid BASE_URL = os.getenv("ALMA_DEMO_URL", "http://127.0.0.1:8080") API_KEY = os.getenv("ALMA_API_KEY", "local-demo-key") def request(method: str, path: str, payload: dict | None = None) -> dict: body = json.dumps(payload).encode("utf-8") if payload is not None else None req = urllib.request.Request( f"{BASE_URL}{path}", data=body, method=method, headers={ "Content-Type": "application/json", "X-ALMA-API-Key": API_KEY, "X-ALMA-Actor": "caregiver", }, ) try: with urllib.request.urlopen(req) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as error: detail = error.read().decode("utf-8") raise RuntimeError(f"{method} {path} failed: {error.code} {detail}") from error def show(title: str, value: dict) -> None: print(f"\n=== {title} ===") print(json.dumps(value, indent=2, ensure_ascii=False)) def main() -> None: resident = request( "POST", "/v1/residents", { "preferred_name": "Maria", "locale": "en-CA", "timezone": "America/Toronto", "quiet_start": "21:00", "quiet_end": "07:00", }, ) show("Resident", resident) contact = request( "POST", f"/v1/residents/{resident['id']}/contacts", { "display_name": "Peter", "relationship": "son", "destination_ref": "mock:contact:peter", "allow_calls": True, "allow_messages": True, "disclosure": ["message_body"], }, ) request( "PUT", f"/v1/residents/{resident['id']}/home-entities/light.living_room", { "entity_id": "light.living_room", "allowed_actions": ["home.light.set"], "constraints": {"states": ["on", "off"]}, }, ) fact = request( "POST", f"/v1/residents/{resident['id']}/facts", { "subject": "Peter", "predicate": "visit.time", "value": {"date": "2026-07-17", "time": "14:00"}, "source_kind": "calendar", "source_reference": "shared-calendar:event-123", "allowed_purposes": ["speak", "display"], }, ) request("POST", f"/v1/facts/{fact['id']}/verify", {}) verified = request( "GET", f"/v1/residents/{resident['id']}/facts/query?predicate=visit.time&purpose=speak", ) show("Verified memory answer source", verified) light = request( "POST", "/v1/actions/proposals", { "request_id": str(uuid.uuid4()), "resident_id": resident["id"], "actor_kind": "resident", "action_type": "home.light.set", "params": {"entity_id": "light.living_room", "state": "on"}, }, ) show("Low-risk local action", light) message = request( "POST", "/v1/actions/proposals", { "request_id": str(uuid.uuid4()), "resident_id": resident["id"], "actor_kind": "resident", "action_type": "message.send", "params": {"contact_id": contact["id"], "body": "I am ready."}, }, ) show("Message pauses for confirmation", message) confirmed = request( "POST", f"/v1/actions/{message['id']}/confirm", {"token": message["confirmation"]["token"], "phrase": "confirm"}, ) show("Confirmed message receipt", confirmed) blocked = request( "POST", "/v1/actions/proposals", { "request_id": str(uuid.uuid4()), "resident_id": resident["id"], "actor_kind": "resident", "action_type": "door.unlock", "params": {"entity_id": "lock.front_door"}, }, ) show("High-risk action remains blocked", blocked) if __name__ == "__main__": main() ======================================================================================== FILE: tests/test_policy.py ======================================================================================== from __future__ import annotations from datetime import datetime, timedelta, timezone import unittest from app.policy import ( ConfirmationSigner, PolicyContext, PolicyEngine, payload_hash, ) class PolicyEngineTests(unittest.TestCase): def setUp(self) -> None: self.policy = PolicyEngine("test-policy") def test_allowlisted_light_is_low_risk_and_immediate(self) -> None: decision = self.policy.evaluate( "home.light.set", {"entity_id": "light.living_room", "state": "on"}, PolicyContext(home_entity_allowed=True), ) self.assertEqual(decision.decision, "allowed") self.assertEqual(decision.risk_class, 1) def test_unlisted_light_is_blocked(self) -> None: decision = self.policy.evaluate( "home.light.set", {"entity_id": "light.living_room", "state": "on"}, PolicyContext(home_entity_allowed=False), ) self.assertEqual(decision.decision, "blocked") self.assertIn("home_entity_not_allowlisted", decision.reason_codes) def test_message_requires_trusted_contact_and_confirmation(self) -> None: decision = self.policy.evaluate( "message.send", {"contact_id": "peter", "body": "I am ready."}, PolicyContext(contact_allowed=True), ) self.assertEqual(decision.decision, "confirmation_required") self.assertEqual(decision.risk_class, 2) self.assertIn("I am ready", decision.confirmation_prompt or "") def test_door_unlock_is_blocked_in_alpha(self) -> None: decision = self.policy.evaluate( "door.unlock", {"entity_id": "lock.front_door"}, PolicyContext(), ) self.assertEqual(decision.decision, "blocked") self.assertEqual(decision.risk_class, 4) def test_financial_transfer_is_always_blocked(self) -> None: decision = self.policy.evaluate( "finance.transfer", {"amount": 1, "destination": "anything"}, PolicyContext(contact_allowed=True, home_entity_allowed=True), ) self.assertEqual(decision.decision, "blocked") self.assertIn("prohibited_in_alpha", decision.reason_codes) def test_confirmation_token_binds_action_payload_and_expiry(self) -> None: signer = ConfirmationSigner("this-is-a-long-test-secret-value", ttl_seconds=120) now = datetime(2026, 7, 17, 12, 0, tzinfo=timezone.utc) digest = payload_hash( "resident-1", "device-1", "message.send", {"contact_id": "peter", "body": "I am ready."}, ) token, _ = signer.issue("action-1", digest, now=now) self.assertTrue(signer.verify(token, "action-1", digest, now=now + timedelta(seconds=30))) self.assertFalse(signer.verify(token, "action-2", digest, now=now + timedelta(seconds=30))) self.assertFalse(signer.verify(token, "action-1", "changed", now=now + timedelta(seconds=30))) self.assertFalse(signer.verify(token, "action-1", digest, now=now + timedelta(seconds=121))) if __name__ == "__main__": unittest.main() ======================================================================================== FILE: tests/test_services.py ======================================================================================== from __future__ import annotations from datetime import datetime, timedelta, timezone from pathlib import Path import tempfile import unittest import uuid from app.adapters import ToolRegistry from app.database import Database from app.policy import ConfirmationSigner, PolicyEngine from app.services import ( ActionService, ContactService, HomeEntityService, MemoryService, ReminderService, ResidentService, ) class ServiceIntegrationTests(unittest.TestCase): def setUp(self) -> None: self.tempdir = tempfile.TemporaryDirectory() self.database = Database(str(Path(self.tempdir.name) / "alma-test.db")) self.database.initialize() self.residents = ResidentService(self.database) self.contacts = ContactService(self.database) self.home = HomeEntityService(self.database) self.memory = MemoryService(self.database) self.reminders = ReminderService(self.database) self.actions = ActionService( self.database, PolicyEngine("test-policy"), ConfirmationSigner("this-is-a-long-service-test-secret", ttl_seconds=120), ToolRegistry(), ) self.resident = self.residents.create( preferred_name="Maria", locale="en-CA", timezone_name="America/Toronto", quiet_start="21:00", quiet_end="07:00", actor_kind="caregiver", ) def tearDown(self) -> None: self.tempdir.cleanup() def test_only_verified_memory_is_returned_for_speaking(self) -> None: fact = self.memory.create_fact( resident_id=self.resident["id"], subject="Peter", predicate="visit.time", value={"time": "14:00", "date": "2026-07-17"}, source_kind="calendar", source_reference="shared-calendar:event-123", created_by="caregiver", confidence=1.0, sensitivity="personal", allowed_purposes=["speak", "display"], valid_from="2026-07-17T00:00:00Z", valid_until="2026-07-17T23:59:59Z", ) before = self.memory.query_verified( resident_id=self.resident["id"], predicate="visit.time", purpose="speak", at=datetime(2026, 7, 17, 12, 0, tzinfo=timezone.utc), ) self.assertEqual(before, []) self.memory.verify_fact(fact["id"], "caregiver") after = self.memory.query_verified( resident_id=self.resident["id"], predicate="visit.time", purpose="speak", at=datetime(2026, 7, 17, 12, 0, tzinfo=timezone.utc), ) self.assertEqual(len(after), 1) self.assertEqual(after[0]["source_reference"], "shared-calendar:event-123") def test_allowed_light_executes_and_writes_receipt(self) -> None: self.home.allow( resident_id=self.resident["id"], entity_id="light.living_room", allowed_actions=["home.light.set"], constraints={}, actor_kind="caregiver", ) action = self.actions.propose( request_id=str(uuid.uuid4()), resident_id=self.resident["id"], device_id=None, actor_kind="resident", action_type="home.light.set", params={"entity_id": "light.living_room", "state": "on"}, ) self.assertEqual(action["state"], "succeeded") self.assertEqual(action["receipt"]["provider"], "mock-home-assistant") def test_message_payload_requires_exact_confirmation(self) -> None: contact = self.contacts.create( resident_id=self.resident["id"], display_name="Peter", relationship="son", destination_ref="mock:contact:peter", allow_calls=True, allow_messages=True, call_window_start="07:00", call_window_end="21:00", disclosure=["message_body"], actor_kind="caregiver", ) action = self.actions.propose( request_id=str(uuid.uuid4()), resident_id=self.resident["id"], device_id=None, actor_kind="resident", action_type="message.send", params={"contact_id": contact["id"], "body": "I am ready."}, ) self.assertEqual(action["state"], "confirmation_required") self.assertNotIn("receipt", action.get("confirmation", {})) completed = self.actions.confirm( action["id"], action["confirmation"]["token"], "confirm", "resident", ) self.assertEqual(completed["state"], "succeeded") self.assertEqual(completed["receipt"]["provider"], "mock-almatalk") def test_reminder_acknowledgement_is_not_adherence_claim(self) -> None: due = datetime.now(timezone.utc) - timedelta(minutes=1) reminder = self.reminders.create( resident_id=self.resident["id"], title="Medication reminder", due_at=due.isoformat(), timezone_name="America/Toronto", recurrence_rule=None, source_reference="caregiver:demo", actor_kind="caregiver", ) self.assertEqual(len(self.reminders.due(self.resident["id"])), 1) acknowledged = self.reminders.acknowledge(reminder["id"], "Resident tapped OK", "resident") self.assertEqual(acknowledged["status"], "acknowledged") if __name__ == "__main__": unittest.main()