diff -ruN '--exclude=.venv' '--exclude=__pycache__' '--exclude=.pytest_cache' '--exclude=.ruff_cache' '--exclude=t*' '--exclude=pc*' /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/.omo/feature-briefs/nutricoach-channel-inbox-v1.5.0.md /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/.omo/feature-briefs/nutricoach-channel-inbox-v1.5.0.md --- /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/.omo/feature-briefs/nutricoach-channel-inbox-v1.5.0.md 1970-01-01 09:00:00.000000000 +0900 +++ /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/.omo/feature-briefs/nutricoach-channel-inbox-v1.5.0.md 2026-08-27 12:26:16.643224345 +0900 @@ -0,0 +1,157 @@ +# NutriCoach Channel Inbox v1.5.0 + +## Goal + +Let customers contact NutriCoach through Telegram Channel Direct Messages while +staff manage every customer in one forum-like channel inbox. A customer must +only see their own conversation. The existing bot-DM and owner operations-room +flows remain the default and remain available for rollback. + +The same candidate adds `nutricoach_multi_customer_v1`: a bounded admission +capability that permits several consenting customers without weakening +per-customer activation, storage, routing, or delivery authority. + +## Release boundary + +- Semantic version: `v1.5.0` compatible feature. +- Capability: `nutricoach_channel_inbox_v1`. +- Capability: `nutricoach_multi_customer_v1`. +- Default: OFF for every profile and customer. +- Development base: frozen v1.1 source in an isolated sibling worktree. +- Integration target: combine the verified v1.4 source and both v1.5 + capabilities into one candidate before the seven-day observation. +- This worktree never modifies the sealed v1.4 candidate, its upgrade package, + live profile data, or the Aug 31 watcher. + +## Actors and authority + +- Customer: an enabled, consenting registry customer whose Telegram user ID + exactly matches `DirectMessagesTopic.user.id`. +- Owner/staff: Telegram channel administrators. Their messages are not accepted + as customer input by the bot. +- Bot: must have the channel right required to manage direct messages. +- Profile authority: exact direct-messages chat ID, parent channel ID, bot ID, + loaded candidate digest, and a derived capability digest. +- Startup calls `getChat` and `getChatMember`; the direct-messages parent and + `can_manage_direct_messages` must match before ingress or outbound routing is + enabled. A config flag or arbitrary digest alone is insufficient. + +## Surfaces + +- Customer surface: a private direct-message conversation with the NutriCoach + channel. +- Owner surface: Telegram's forum-like channel direct-message inbox, one topic + per customer. +- Existing surface: bot private DM and the owner-only operations group remain + unchanged while the capability is OFF. + +## State and data + +1. `OFF`: channel-DM updates are not routed into NutriCoach. +2. `ARMED`: config, channel ID, bot authority, registry, and capability receipt + are valid; customers still use their existing bot-DM routes. +3. `ROUTED`: a permission-sealed activation has changed one customer's + canonical registry route to the exact channel direct-messages chat/topic. +4. `DISABLED`: the flag is OFF. A separately approved registry rollback returns + the customer to the archived bot-DM route. + +The runtime adds no mutable routing sidecar. The existing customer registry +remains the sole route authority. Topic discovery and the proposed registry +mutation are activation evidence outside the runtime; they require a separate +one-use permission seal before `ROUTED`. + +The shipped migration core is pure: it canonicalizes registry bytes, binds +before/after hashes, derives an exact approval phrase, applies only the expected +customer route in memory, and proves byte-identical rollback. A later live +controller must additionally stop the Gateway, verify no pending customer +workflow, persist atomically, restart, and smoke-test. + +## Bounded multi-customer admission + +- A legacy registry without the capability remains limited to one enabled + customer. +- A versioned, frozen admission policy sets an explicit limit from 2 through + 20. The initial live pilot uses 5. +- Multiple disabled drafts remain allowed while the capability is OFF. + Enabling each customer still requires that customer's consent, onboarding, + owner review, membership, readiness, and activation transaction. +- Activation receipts are durable per customer. Adding or activating another + customer cannot invalidate an unrelated customer's receipt. Changing a + customer's projection, the owner, or the admission policy fails closed. +- Activations remain serialized by the profile authority lock. Crash recovery + reconciles one in-flight activation without sharing receipts or data roots. +- When multiple customers are enabled, the production operator console + requires an explicit `customer_key`; it never selects the first customer. +- Every customer retains a distinct registry key, Telegram route, data root, + wizard state, event store, schedule fence, draft, delivery claim, and + customer-surface receipt. +- Expanding capacity is a separately permission-sealed registry migration. + Removing the policy is allowed only after all but one customer are disabled; + rollback never silently disables customers. + +## Inbound contract + +An update is accepted as customer input only when all are true: + +- capability is enabled and receipt-bound; +- `chat.is_direct_messages` is exactly true; +- chat ID equals the configured direct-messages chat ID; +- `message.direct_messages_topic` exists; +- topic ID is a positive Telegram-safe integer; +- `topic.user` exists; +- `message.from_user.id == topic.user.id`; +- exactly one enabled, consenting registry customer has that exact + user/chat/topic route; +- no conflicting registry route exists. + +The normalized source uses the direct-messages chat as `chat_id`, the topic as +`thread_id`, and the topic user as `user_id`. This preserves existing customer +resolution and Telegram's `direct_messages_topic_id` outbound routing. + +Admin/channel-originated messages, anonymous senders, unknown customers, +ambiguous registry entries, wrong channels, malformed topics, and conflicts are +ignored or fail closed before session creation or persistence. + +## Outbound contract + +- A live reply stays in the triggering direct-message topic. +- A scheduled send may use the registry topic only when the capability, + receipt, and registry identity still match. +- Otherwise delivery falls back to the existing bot-DM route. +- `message_thread_id` and `direct_messages_topic_id` are never sent together. +- No cross-customer topic fallback is allowed. + +## Privacy and observability + +- Customers never join the owner operations group. +- Telegram supplies customer isolation through Channel Direct Messages. +- Logs contain customer keys and typed reason codes, not message bodies or raw + user IDs. +- Durable nutrition events remain authoritative; the Telegram inbox is an + operational view. + +## Migration and rollback + +- No historical message migration. +- Existing customers remain on bot DM until a channel topic is proven and a + permission-sealed registry migration is committed. +- Turning the capability OFF prevents channel-inbox ingress. Rolling a customer + registry route back to its archived bot-DM address restores automated sends. +- Removing the v1.5 overlay restores v1.4 behavior without rewriting customer + events, registry entries, or historical receipts. + +## Verification + +- RED/GREEN tests for inbound identity, admin-loop prevention, route collision, + disabled behavior, synthetic scheduled routing, and bot-DM fallback. +- RED/GREEN tests for legacy one-customer behavior, bounded capacity, two + independently activated customers, per-customer receipt persistence, + cross-customer route/storage isolation, concurrent activation serialization, + restart recovery, capacity rollback, and explicit console selection. +- Existing Telegram topic, reply-mode, delivery, rich-message, nutrition, and + operations-room tests remain green. +- Strict type, Ruff, compile, no-excuse, privacy-negative, and module-size gates. +- Real `python-telegram-bot` objects exercise a customer message, owner reply isolation, + a scheduled send, a wrong-topic attack, and rollback to bot DM. +- Live channel setup, customer migration, and bot-right changes require a later + permission-sealed activation and are out of scope here. diff -ruN '--exclude=.venv' '--exclude=__pycache__' '--exclude=.pytest_cache' '--exclude=.ruff_cache' '--exclude=t*' '--exclude=pc*' /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/.pytest-cache/gw-adapter-guard-37bd2831f3adfc61 /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/.pytest-cache/gw-adapter-guard-37bd2831f3adfc61 --- /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/.pytest-cache/gw-adapter-guard-37bd2831f3adfc61 1970-01-01 09:00:00.000000000 +0900 +++ /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/.pytest-cache/gw-adapter-guard-37bd2831f3adfc61 2026-08-27 14:47:01.223143271 +0900 @@ -0,0 +1 @@ +clean \ 파일 끝 개행 문자 없음 diff -ruN '--exclude=.venv' '--exclude=__pycache__' '--exclude=.pytest_cache' '--exclude=.ruff_cache' '--exclude=t*' '--exclude=pc*' /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/.pytest-cache/gw-adapter-guard-61cef65fd951e18a /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/.pytest-cache/gw-adapter-guard-61cef65fd951e18a --- /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/.pytest-cache/gw-adapter-guard-61cef65fd951e18a 2026-08-26 02:57:46.095635204 +0900 +++ /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/.pytest-cache/gw-adapter-guard-61cef65fd951e18a 1970-01-01 09:00:00.000000000 +0900 @@ -1 +0,0 @@ -clean \ 파일 끝 개행 문자 없음 diff -ruN '--exclude=.venv' '--exclude=__pycache__' '--exclude=.pytest_cache' '--exclude=.ruff_cache' '--exclude=t*' '--exclude=pc*' /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/channel_inbox_migration.py /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/channel_inbox_migration.py --- /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/channel_inbox_migration.py 1970-01-01 09:00:00.000000000 +0900 +++ /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/channel_inbox_migration.py 2026-08-27 13:22:14.476070335 +0900 @@ -0,0 +1,238 @@ +# Copyright (c) 2026 Nous Research +"""Pure canonical registry migration for a channel inbox route.""" + +from __future__ import annotations + +from dataclasses import asdict, replace +import hashlib +import hmac +import json +import re + +from pydantic import JsonValue, TypeAdapter, ValidationError + +from checkin_cli.channel_inbox_migration_models import ( + ChannelInboxMigrationError, + ChannelInboxMigrationProposal, + ChannelInboxMigrationRequest, + MigrationRejectReason, + TelegramRoute, +) +from checkin_cli.customer_coaching import CustomerRegistryError, RegistryDocument +from gateway.platforms.telegram_channel_inbox_config import ( + channel_inbox_capability_digest, +) + +_DIGEST = re.compile(r"[a-f0-9]{64}") +_JSON_OBJECT = TypeAdapter(dict[str, JsonValue]) +_JSON_OBJECTS = TypeAdapter(list[dict[str, JsonValue]]) +_MAX_TOPIC_ID = (1 << 52) - 1 + + +def _sha256(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def _canonical_registry(registry: dict[str, JsonValue]) -> bytes: + return (json.dumps(registry, ensure_ascii=False, indent=2) + "\n").encode() + + +def _parse_registry( + payload: bytes, +) -> tuple[dict[str, JsonValue], list[dict[str, JsonValue]]]: + try: + document = RegistryDocument.model_validate_json(payload) + registry = _JSON_OBJECT.validate_json(payload) + customers = _JSON_OBJECTS.validate_python(registry.get("customers")) + except (CustomerRegistryError, ValidationError) as exc: + raise ChannelInboxMigrationError(MigrationRejectReason.REGISTRY) from exc + if document.registry_mode != "ordinary_v1": + raise ChannelInboxMigrationError(MigrationRejectReason.REGISTRY) + if _canonical_registry(registry) != payload: + raise ChannelInboxMigrationError(MigrationRejectReason.REGISTRY) + return registry, customers + + +def _route(value: JsonValue) -> TelegramRoute: + try: + route = _JSON_OBJECT.validate_python(value) + except ValidationError as exc: + raise ChannelInboxMigrationError(MigrationRejectReason.IDENTITY) from exc + fields = tuple( + str(route.get(name, "")).strip() for name in ("user_id", "chat_id", "topic_id") + ) + if any(not field for field in fields): + raise ChannelInboxMigrationError(MigrationRejectReason.IDENTITY) + return TelegramRoute(*fields) + + +def _canonical_integer(value: str, reason: MigrationRejectReason) -> int: + try: + parsed = int(value) + except ValueError as exc: + raise ChannelInboxMigrationError(reason) from exc + if parsed == 0 or value != str(parsed): + raise ChannelInboxMigrationError(reason) + return parsed + + +def _proposal_digest(proposal: ChannelInboxMigrationProposal) -> str: + payload = { + "after_sha256": proposal.after_sha256, + "before_sha256": proposal.before_sha256, + "authority": asdict(proposal.authority), + "capability_digest": proposal.capability_digest, + "customer_key": proposal.customer_key, + "new_route": asdict(proposal.new_route), + "old_route": asdict(proposal.old_route), + "schema": "nutricoach-channel-inbox-route-migration-v1", + } + canonical = json.dumps( + payload, ensure_ascii=True, separators=(",", ":"), sort_keys=True + ) + return hashlib.sha256(canonical.encode("ascii")).hexdigest() + + +def _validate_digest(value: str) -> None: + if _DIGEST.fullmatch(value) is None: + raise ChannelInboxMigrationError(MigrationRejectReason.DIGEST) + + +def _validate_proposal(proposal: ChannelInboxMigrationProposal) -> None: + _validate_digest(proposal.candidate_digest) + _validate_digest(proposal.capability_digest) + if proposal.capability_digest != channel_inbox_capability_digest( + proposal.authority, + ): + raise ChannelInboxMigrationError(MigrationRejectReason.AUTHORITY) + expected = _proposal_digest(proposal) + if not hmac.compare_digest(expected, proposal.proposal_digest): + raise ChannelInboxMigrationError(MigrationRejectReason.PROPOSAL) + + +def _migrated_bytes( + registry_bytes: bytes, + *, + customer_key: str, + new_route: TelegramRoute, +) -> tuple[bytes, TelegramRoute]: + registry, customers = _parse_registry(registry_bytes) + matches = [row for row in customers if row.get("customer_key") == customer_key] + if len(matches) != 1: + raise ChannelInboxMigrationError(MigrationRejectReason.CUSTOMER) + customer = matches[0] + if customer.get("enabled") is not True: + raise ChannelInboxMigrationError(MigrationRejectReason.ENABLED) + try: + consent = _JSON_OBJECT.validate_python(customer.get("ai_processing_consent")) + except ValidationError as exc: + raise ChannelInboxMigrationError(MigrationRejectReason.CONSENT) from exc + if consent.get("granted") is not True: + raise ChannelInboxMigrationError(MigrationRejectReason.CONSENT) + old_route = _route(customer.get("telegram")) + if old_route.user_id != new_route.user_id: + raise ChannelInboxMigrationError(MigrationRejectReason.IDENTITY) + owner_route = _route(registry.get("owner")) + occupied = { + owner_route, + *(_route(row.get("telegram")) for row in customers if row is not customer), + } + if new_route in occupied or (new_route.chat_id, new_route.topic_id) in { + (route.chat_id, route.topic_id) for route in occupied + }: + raise ChannelInboxMigrationError(MigrationRejectReason.COLLISION) + customer["telegram"] = asdict(new_route) + customers_json: list[JsonValue] = [customer for customer in customers] + registry["customers"] = customers_json + return _canonical_registry(registry), old_route + + +def plan_registry_migration( + registry_bytes: bytes, + request: ChannelInboxMigrationRequest, +) -> ChannelInboxMigrationProposal: + """Build an immutable migration proposal without writing files.""" + _validate_digest(request.authority.candidate_digest) + _validate_digest(request.capability_digest) + if request.capability_digest != channel_inbox_capability_digest( + request.authority, + ): + raise ChannelInboxMigrationError(MigrationRejectReason.AUTHORITY) + _ = _canonical_integer( + request.authority.direct_messages_chat_id, + MigrationRejectReason.IDENTITY, + ) + topic_id = _canonical_integer( + request.direct_messages_topic_id, + MigrationRejectReason.TOPIC, + ) + if topic_id <= 0 or topic_id > _MAX_TOPIC_ID: + raise ChannelInboxMigrationError(MigrationRejectReason.TOPIC) + new_route = TelegramRoute( + request.telegram_user_id.strip(), + request.authority.direct_messages_chat_id, + request.direct_messages_topic_id, + ) + migrated, old_route = _migrated_bytes( + registry_bytes, + customer_key=request.customer_key, + new_route=new_route, + ) + before_sha256 = _sha256(registry_bytes) + after_sha256 = _sha256(migrated) + proposal = ChannelInboxMigrationProposal( + request.customer_key, + before_sha256, + after_sha256, + old_route, + new_route, + request.authority, + request.capability_digest, + "", + ) + return replace(proposal, proposal_digest=_proposal_digest(proposal)) + + +def apply_registry_migration( + registry_bytes: bytes, + proposal: ChannelInboxMigrationProposal, + approval: str, +) -> bytes: + """Apply an exact approved proposal in memory.""" + _validate_proposal(proposal) + if not hmac.compare_digest( + approval.encode(), + proposal.approval_phrase.encode(), + ): + raise ChannelInboxMigrationError(MigrationRejectReason.APPROVAL) + if _sha256(registry_bytes) != proposal.before_sha256: + raise ChannelInboxMigrationError(MigrationRejectReason.BEFORE) + migrated, old_route = _migrated_bytes( + registry_bytes, + customer_key=proposal.customer_key, + new_route=proposal.new_route, + ) + if old_route != proposal.old_route or _sha256(migrated) != proposal.after_sha256: + raise ChannelInboxMigrationError(MigrationRejectReason.AFTER) + return migrated + + +def rollback_registry_migration( + registry_bytes: bytes, + proposal: ChannelInboxMigrationProposal, +) -> bytes: + """Rollback only the exact migrated registry bytes.""" + _validate_proposal(proposal) + if _sha256(registry_bytes) != proposal.after_sha256: + raise ChannelInboxMigrationError(MigrationRejectReason.AFTER) + rolled_back, old_route = _migrated_bytes( + registry_bytes, + customer_key=proposal.customer_key, + new_route=proposal.old_route, + ) + if ( + old_route != proposal.new_route + or _sha256(rolled_back) != proposal.before_sha256 + ): + raise ChannelInboxMigrationError(MigrationRejectReason.BEFORE) + return rolled_back diff -ruN '--exclude=.venv' '--exclude=__pycache__' '--exclude=.pytest_cache' '--exclude=.ruff_cache' '--exclude=t*' '--exclude=pc*' /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/channel_inbox_migration_models.py /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/channel_inbox_migration_models.py --- /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/channel_inbox_migration_models.py 1970-01-01 09:00:00.000000000 +0900 +++ /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/channel_inbox_migration_models.py 2026-08-27 08:55:39.547926625 +0900 @@ -0,0 +1,94 @@ +# Copyright (c) 2026 Nous Research +"""Typed identities for a sealed channel-inbox registry migration.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from typing import override + +from gateway.platforms.telegram_channel_inbox_config import ChannelInboxAuthority + + +class MigrationRejectReason(StrEnum): + """Stable fail-closed migration reasons.""" + + AFTER = "after" + APPROVAL = "approval" + AUTHORITY = "authority" + BEFORE = "before" + COLLISION = "collision" + CONSENT = "consent" + CUSTOMER = "customer" + DIGEST = "digest" + ENABLED = "enabled" + IDENTITY = "identity" + PROPOSAL = "proposal" + REGISTRY = "registry" + TOPIC = "topic" + + +@dataclass(frozen=True, slots=True) +class ChannelInboxMigrationError(ValueError): + """The registry migration failed before external persistence.""" + + reason: MigrationRejectReason + + @override + def __str__(self) -> str: + """Return the machine-consumed reason. + + Returns: + Stable rejection reason. + + """ + return self.reason.value + + +@dataclass(frozen=True, slots=True) +class TelegramRoute: + """Exact Telegram delivery address.""" + + user_id: str + chat_id: str + topic_id: str + + +@dataclass(frozen=True, slots=True) +class ChannelInboxMigrationRequest: + """Owner-requested route transition bound to one candidate.""" + + customer_key: str + telegram_user_id: str + direct_messages_topic_id: str + authority: ChannelInboxAuthority + capability_digest: str + + +@dataclass(frozen=True, slots=True) +class ChannelInboxMigrationProposal: + """Immutable input to a later permission-sealed live controller.""" + + customer_key: str + before_sha256: str + after_sha256: str + old_route: TelegramRoute + new_route: TelegramRoute + authority: ChannelInboxAuthority + capability_digest: str + proposal_digest: str + + @property + def candidate_digest(self) -> str: + """Return the candidate bound by the Telegram authority.""" + return self.authority.candidate_digest + + @property + def approval_phrase(self) -> str: + """Return the exact one-use owner approval phrase. + + Returns: + Exact phrase bound to this proposal. + + """ + return f"AUTHORIZE NUTRICOACH CHANNEL INBOX ROUTE V1 {self.proposal_digest}" diff -ruN '--exclude=.venv' '--exclude=__pycache__' '--exclude=.pytest_cache' '--exclude=.ruff_cache' '--exclude=t*' '--exclude=pc*' /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/customer_admin.py /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/customer_admin.py --- /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/customer_admin.py 2026-08-25 22:21:25.335830680 +0900 +++ /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/customer_admin.py 2026-08-27 14:43:08.890978800 +0900 @@ -10,15 +10,16 @@ import uuid import fcntl import stat +from collections.abc import Iterator, Mapping from contextlib import contextmanager from contextvars import ContextVar from types import MappingProxyType from dataclasses import dataclass from datetime import date, datetime, time, timedelta, timezone from pathlib import Path -from typing import Callable, Iterator, Mapping +from typing import Callable, NoReturn, TypeVar from threading import local -from pydantic import ValidationError +from pydantic import JsonValue, TypeAdapter, ValidationError from zoneinfo import ZoneInfo from checkin_cli.activation_token_rotation_policy import ( @@ -49,6 +50,10 @@ ) from checkin_cli.weekly_operations_schedule_host_models_r4 import CustomerScheduleError from checkin_cli.models import OperatorTask, PaymentKind, PaymentMethod, RecordResult +from checkin_cli.multi_customer_admission_migration_models import ( + AdmissionMigrationError, + AdmissionMigrationProposal, +) from checkin_cli.nutrition_readiness import ( NUTRITION_READINESS_REASON_CODES, NutritionReadinessError, @@ -714,7 +719,7 @@ try: document.validate_pilot_customer_boundary(customer) except CustomerRegistryError as exc: - raise CustomerAdminError("pilot supports only one external customer") from exc + raise CustomerAdminError("enabled customer capacity exceeded") from exc payload = document.model_dump(mode="json") customers = payload["customers"] if not isinstance(customers, list): @@ -891,6 +896,92 @@ ) +def commit_multi_customer_admission_migration( + profile_root: Path, + proposal: AdmissionMigrationProposal, + approval_phrase: str, +) -> str: + """Atomically enable bounded admission and preserve current authority.""" + from checkin_cli.multi_customer_admission_migration import ( + apply_admission_migration, + ) + + root = _resolve_profile_root(profile_root) + registry_path = _resolve_registry_path(root) + with profile_authority_lock(root): + _recover_activation_journal_locked(root, registry_path) + before = registry_path.read_bytes() + document = _read_profile_registry(registry_path, root) + _validate_enabled_activation_receipts(root, registry_path, document) + try: + after = apply_admission_migration( + before, + proposal, + approval_phrase, + ) + migrated = RegistryDocument.model_validate_json(after) + except (AdmissionMigrationError, TypeError, ValueError) as exc: + raise CustomerAdminError( + "multi-customer admission migration rejected" + ) from exc + + receipt_payloads: dict[Path, dict[str, JsonValue]] = {} + for spec in document.customers: + if not spec.enabled: + continue + journal = _read_activation_journal(_activation_journal_path(root)) + if journal is None: + raise CustomerAdminError( + "committed activation receipt is missing or stale" + ) + _ = _validate_committed_activation_payload( + root, + registry_path, + document, + spec, + journal, + ) + receipt = _build_customer_activation_receipt( + journal, + migrated, + spec, + ) + _ = _validate_committed_activation_payload( + root, + registry_path, + migrated, + spec, + receipt, + ) + receipt_payloads[_activation_receipt_path(root, spec.customer_key)] = ( + receipt + ) + + receipt_before = {path: _read_optional_bytes(path) for path in receipt_payloads} + try: + for path, receipt in receipt_payloads.items(): + _write_activation_journal(path, receipt) + _atomic_write_bytes(registry_path, after) + _validate_enabled_activation_receipts( + root, + registry_path, + migrated, + ) + except (CustomerAdminError, OSError, TypeError, ValueError) as exc: + try: + _restore_bytes(registry_path, before) + for path, previous in receipt_before.items(): + _restore_bytes(path, previous) + except (CustomerAdminError, OSError) as rollback_error: + raise CustomerAdminError( + "multi-customer admission rollback failed" + ) from rollback_error + raise CustomerAdminError( + "multi-customer admission migration failed" + ) from exc + return proposal.after_sha256 + + def _activate_customer_locked( profile_root: Path, data_root: Path, @@ -948,7 +1039,12 @@ try: document.validate_pilot_customer_boundary(enabled_spec) except (CustomerRegistryError, ValueError) as exc: - raise CustomerAdminError("pilot supports only one external customer") from exc + message = ( + "pilot supports only one external customer" + if document.admission_policy is None + else "enabled customer capacity exceeded" + ) + raise CustomerAdminError(message) from exc try: spec.validate_local_activation_prerequisites() @@ -1029,6 +1125,14 @@ registry_before = registry_path.read_bytes() audit_path = _activation_audit_path(root) audit_before = _read_optional_bytes(audit_path) + receipt_path = ( + _activation_receipt_path(root, customer_id) + if activated_document.admission_policy is not None + else None + ) + receipt_before = ( + _read_optional_bytes(receipt_path) if receipt_path is not None else None + ) transaction_id = uuid.uuid4().hex registry_sha256 = _document_fingerprint(activated_document) recorded_at = datetime.now(timezone.utc).isoformat() @@ -1062,6 +1166,11 @@ "prepared_at": recorded_at, "nutrition_activation_receipt": nutrition_activation_receipt, **membership_bindings, + **( + {"receipt_committed_at": recorded_at} + if activated_document.admission_policy is not None + else {} + ), } journal_path = _activation_journal_path(root) _write_activation_journal(journal_path, pending) @@ -1096,15 +1205,24 @@ or written_audit_state != "present" ): raise CustomerAdminError("activation side effects could not be verified") - _write_activation_journal( - journal_path, - { - **pending, - "state": "committed", - "recovery_required": False, - "committed_at": datetime.now(timezone.utc).isoformat(), - }, - ) + committed = _JSON_OBJECT.validate_python({ + **pending, + "state": "committed", + "recovery_required": False, + "committed_at": ( + pending.get("receipt_committed_at") + or datetime.now(timezone.utc).isoformat() + ), + }) + _write_activation_journal(journal_path, committed) + if activated_document.admission_policy is not None: + _ = _materialize_customer_activation_receipt( + root, + registry_path, + activated_document, + enabled_spec, + committed, + ) except Exception as exc: try: _rollback_activation( @@ -1114,6 +1232,8 @@ audit_before, journal_path, pending, + receipt_path=receipt_path, + receipt_before=receipt_before, ) except CustomerAdminError as rollback_error: raise rollback_error from exc @@ -1229,6 +1349,11 @@ _ACTIVATION_JOURNAL_FILE = "customer-activation-journal.json" _ACTIVATION_AUDIT_FILE = "customer-activation-audit.jsonl" +_ACTIVATION_RECEIPT_DIRECTORY = "customer-activation-receipts" +_CUSTOMER_ACTIVATION_RECEIPT_VERSION = 4 +_JSON_VALUE: TypeAdapter[JsonValue] = TypeAdapter(JsonValue) +_JSON_OBJECT = TypeAdapter(dict[str, JsonValue]) +_JournalValue = TypeVar("_JournalValue") def _activation_journal_path(profile_root: Path) -> Path: @@ -1239,6 +1364,11 @@ return profile_root / "data" / _ACTIVATION_AUDIT_FILE +def _activation_receipt_path(profile_root: Path, customer_key: str) -> Path: + key = _require_customer_key(customer_key) + return profile_root / "data" / _ACTIVATION_RECEIPT_DIRECTORY / f"{key}.json" + + def _json_digest(value: object) -> str: payload = json.dumps( value, @@ -1324,7 +1454,7 @@ _atomic_write_bytes(path, previous) -def _read_activation_journal(path: Path) -> dict[str, object] | None: +def _read_activation_journal(path: Path) -> dict[str, JsonValue] | None: if path.is_symlink(): raise CustomerAdminError("activation journal symlinks are not allowed") if not path.exists(): @@ -1332,15 +1462,18 @@ if not path.is_file(): raise CustomerAdminError("activation journal must be a regular file") try: - payload = json.loads(path.read_text(encoding="utf-8")) - except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + payload = _JSON_VALUE.validate_json(path.read_bytes()) + except (OSError, ValidationError) as exc: raise CustomerAdminError("activation journal is invalid") from exc if not isinstance(payload, dict): raise CustomerAdminError("activation journal must be a JSON object") return payload -def _write_activation_journal(path: Path, payload: dict[str, object]) -> None: +def _write_activation_journal( + path: Path, + payload: Mapping[str, _JournalValue], +) -> None: state = payload.get("state") if state not in {"prepared", "committed", "abandoned"}: raise CustomerAdminError("activation journal state is invalid") @@ -1454,6 +1587,9 @@ audit_before: bytes | None, journal_path: Path, pending: dict[str, object], + *, + receipt_path: Path | None = None, + receipt_before: bytes | None = None, ) -> None: failures: list[dict[str, str]] = [] registry_error: BaseException | None = None @@ -1498,6 +1634,29 @@ ) ) + if receipt_path is not None: + receipt_error: BaseException | None = None + try: + _restore_bytes(receipt_path, receipt_before) + except (CustomerAdminError, OSError) as exc: + receipt_error = exc + try: + receipt_after = _read_optional_bytes(receipt_path) + except (CustomerAdminError, OSError) as exc: + receipt_after = object() + receipt_error = receipt_error or exc + if receipt_after != receipt_before: + failures.append( + _activation_failure_detail( + "customer_receipt", + "restore", + receipt_error + or CustomerAdminError( + "customer receipt bytes do not match the prepared snapshot" + ), + ) + ) + if failures: _record_activation_recovery_required(journal_path, pending, failures) raise CustomerAdminError("activation recovery required") @@ -1615,7 +1774,28 @@ state = journal.get("state") if journal.get("version") not in {1, 2, 3}: raise CustomerAdminError("activation journal version is invalid") - if state in {"committed", "abandoned"}: + if state == "committed": + document = _read_profile_registry(registry_path, profile_root) + if document.admission_policy is not None: + customer_id = journal.get("customer_id") + spec = next( + ( + item + for item in document.customers + if item.customer_key == customer_id + ), + None, + ) + if spec is None: + raise CustomerAdminError( + "committed activation receipt is missing or stale" + ) + if spec.enabled: + _ = _materialize_customer_activation_receipt( + profile_root, registry_path, document, spec, journal + ) + return + if state == "abandoned": return if state not in {"prepared", "pending", "recovery_required"}: raise CustomerAdminError("activation journal has an unknown state") @@ -1692,7 +1872,7 @@ component: str, operation: str, error: BaseException, - ) -> None: + ) -> NoReturn: failure = _activation_failure_detail(component, operation, error) _record_activation_recovery_required(journal_path, journal, [failure]) raise CustomerAdminError("pending activation recovery required") from error @@ -1713,16 +1893,39 @@ require_recovery("audit", "inspect", exc) def terminalize(state_name: str) -> None: + terminal = { + **journal, + "state": state_name, + "recovery_required": False, + f"{state_name}_at": ( + journal.get("receipt_committed_at") + if state_name == "committed" + and isinstance(journal.get("receipt_committed_at"), str) + else datetime.now(timezone.utc).isoformat() + ), + } try: - _write_activation_journal( - journal_path, - { - **journal, - "state": state_name, - "recovery_required": False, - f"{state_name}_at": datetime.now(timezone.utc).isoformat(), - }, - ) + _write_activation_journal(journal_path, terminal) + if state_name == "committed" and current.admission_policy is not None: + committed_spec = next( + ( + item + for item in current.customers + if item.customer_key == customer_id + ), + None, + ) + if committed_spec is None or not committed_spec.enabled: + raise CustomerAdminError( + "committed activation customer is missing or stale" + ) + _ = _materialize_customer_activation_receipt( + profile_root, + registry_path, + current, + committed_spec, + terminal, + ) except Exception as exc: require_recovery("journal", f"record {state_name} state", exc) @@ -1826,33 +2029,90 @@ return False -def _require_committed_activation_receipt( +def _audit_receipt_binds_customer_payload( + audit_path: Path, + receipt: Mapping[str, JsonValue], +) -> bool: + transaction_id = receipt.get("transaction_id") + audit_record_sha256 = receipt.get("audit_record_sha256") + if not isinstance(transaction_id, str) or not isinstance(audit_record_sha256, str): + return False + try: + lines = audit_path.read_bytes().splitlines() + except OSError: + return False + matching: dict[str, JsonValue] | None = None + for line in lines: + if not line.strip(): + continue + try: + row = _JSON_VALUE.validate_json(line) + except ValidationError: + return False + if not isinstance(row, dict) or row.get("transaction_id") != transaction_id: + continue + if matching is not None or _json_digest(row) != audit_record_sha256: + return False + matching = row + if matching is None: + return False + bound_fields = { + "customer_id", + "registry_path", + "data_root", + "checklist_evidence_path", + } + source_version = receipt.get("source_journal_version") + if source_version in {2, 3}: + bound_fields.add("nutrition_activation_receipt") + if source_version == 3: + bound_fields.update({ + "staff_membership_evidence_path", + "staff_membership_evidence_sha256", + "staff_chat_inventory_sha256", + "membership_subscription_epoch_id", + }) + return all(matching.get(field) == receipt.get(field) for field in bound_fields) + + +def _validate_committed_activation_payload( profile_root: Path, registry_path: Path, document: RegistryDocument, spec: CustomerSpec, + receipt: Mapping[str, JsonValue], ) -> str: - journal = _read_activation_journal(_activation_journal_path(profile_root)) - if journal is None or journal.get("state") != "committed": + if receipt.get("state") != "committed": raise CustomerAdminError("committed activation receipt is missing or stale") - version = journal.get("version") - if version not in {1, 2, 3} or journal.get("customer_id") != spec.customer_key: - raise CustomerAdminError("committed activation receipt is missing or stale") - if journal.get("recovery_required") is True: + version = receipt.get("version") + source_version = ( + receipt.get("source_journal_version") + if version == _CUSTOMER_ACTIVATION_RECEIPT_VERSION + else version + ) + if ( + version not in {1, 2, 3, _CUSTOMER_ACTIVATION_RECEIPT_VERSION} + or source_version not in {1, 2, 3} + or receipt.get("customer_id") != spec.customer_key + or receipt.get("recovery_required") is True + ): raise CustomerAdminError("committed activation receipt is missing or stale") - transaction_id = journal.get("transaction_id") - registry_sha256 = journal.get("registry_sha256") - audit_record_sha256 = journal.get("audit_record_sha256") + transaction_id = receipt.get("transaction_id") + registry_sha256 = receipt.get("registry_sha256") + audit_record_sha256 = receipt.get("audit_record_sha256") if ( not isinstance(transaction_id, str) or not transaction_id - or not isinstance(registry_sha256, str) + or ( + version != _CUSTOMER_ACTIVATION_RECEIPT_VERSION + and not isinstance(registry_sha256, str) + ) or not isinstance(audit_record_sha256, str) ): raise CustomerAdminError("committed activation receipt is missing or stale") - if version in {2, 3}: - receipt = journal.get("nutrition_activation_receipt") - if not isinstance(receipt, dict): + if source_version in {2, 3}: + nutrition_receipt = receipt.get("nutrition_activation_receipt") + if not isinstance(nutrition_receipt, dict): raise CustomerAdminError( "nutrition activation v2 receipt is missing or stale" ) @@ -1875,19 +2135,19 @@ raise CustomerAdminError( "nutrition activation v2 receipt is missing or stale" ) from exc - if receipt != expected_receipt: + if nutrition_receipt != expected_receipt: raise CustomerAdminError( "nutrition activation v2 receipt is missing or stale" ) - if version == 3: - membership_path = journal.get("staff_membership_evidence_path") + if source_version == 3: + membership_path = receipt.get("staff_membership_evidence_path") required_membership_fields = ( "staff_membership_evidence_sha256", "staff_chat_inventory_sha256", "membership_subscription_epoch_id", ) if not isinstance(membership_path, str) or any( - not isinstance(journal.get(field), str) or not journal.get(field) + not isinstance(receipt.get(field), str) or not receipt.get(field) for field in required_membership_fields ): raise CustomerAdminError( @@ -1896,25 +2156,25 @@ resolved_membership = _resolve_journal_path( membership_path, profile_root, "staff_membership_evidence_path" ) - try: - membership_bytes = resolved_membership.read_bytes() - membership_digest = hashlib.sha256(membership_bytes).hexdigest() - membership_payload = json.loads(membership_bytes) - except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: - raise CustomerAdminError( - "committed staff membership evidence is missing or stale" - ) from exc - if ( - not isinstance(membership_payload, dict) - or membership_digest != journal.get("staff_membership_evidence_sha256") - or membership_payload.get("staff_chat_inventory_sha256") - != journal.get("staff_chat_inventory_sha256") - or membership_payload.get("subscription_epoch_id") - != journal.get("membership_subscription_epoch_id") - ): - raise CustomerAdminError( - "committed staff membership evidence is missing or stale" - ) + if version != _CUSTOMER_ACTIVATION_RECEIPT_VERSION: + try: + membership_bytes = resolved_membership.read_bytes() + membership_digest = hashlib.sha256(membership_bytes).hexdigest() + membership_payload = _JSON_OBJECT.validate_json(membership_bytes) + except (OSError, ValidationError) as exc: + raise CustomerAdminError( + "committed staff membership evidence is missing or stale" + ) from exc + if ( + membership_digest != receipt.get("staff_membership_evidence_sha256") + or membership_payload.get("staff_chat_inventory_sha256") + != receipt.get("staff_chat_inventory_sha256") + or membership_payload.get("subscription_epoch_id") + != receipt.get("membership_subscription_epoch_id") + ): + raise CustomerAdminError( + "committed staff membership evidence is missing or stale" + ) else: manifest_path = ( profile_root @@ -1925,15 +2185,22 @@ ) try: validate_profile_path(manifest_path, profile_root) - manifest = read_private_json(manifest_path) - except (OSError, ValueError) as exc: + manifest = _JSON_OBJECT.validate_python(read_private_json(manifest_path)) + except (OSError, ValueError, ValidationError) as exc: raise CustomerAdminError( "legacy activation authority manifest is missing or stale" ) from exc - if not isinstance(manifest, dict) or not validate_legacy_activation_authority( - manifest, + legacy_digest = ( + receipt.get("legacy_activation_receipt_sha256") + if version == _CUSTOMER_ACTIVATION_RECEIPT_VERSION + else _json_digest(receipt) + ) + if not isinstance( + legacy_digest, str + ) or not validate_legacy_activation_authority( + {key: value for key, value in manifest.items() if isinstance(value, str)}, customer_key=spec.customer_key, - activation_receipt_digest=_json_digest(journal), + activation_receipt_digest=legacy_digest, registry_projection_digest=customer_nutrition_projection_digest( spec.model_dump(mode="json") ), @@ -1944,19 +2211,23 @@ ) try: receipt_registry = _resolve_journal_path( - journal.get("registry_path"), - profile_root, - "registry_path", + receipt.get("registry_path"), profile_root, "registry_path" ) receipt_data = _resolve_journal_path( - journal.get("data_root"), - profile_root, - "data_root", + receipt.get("data_root"), profile_root, "data_root" ) + if version == _CUSTOMER_ACTIVATION_RECEIPT_VERSION: + checklist_value = receipt.get("checklist_evidence_path") + if not isinstance(checklist_value, str) or not checklist_value: + raise CustomerAdminError("activation journal checklist path is invalid") + checklist_path = Path(checklist_value) + if checklist_path.is_symlink(): + raise CustomerAdminError( + "activation journal checklist path symlinks are not allowed" + ) + _ = checklist_path.resolve() receipt_audit = _resolve_journal_path( - journal.get("audit_path"), - profile_root, - "audit_path", + receipt.get("audit_path"), profile_root, "audit_path" ) except CustomerAdminError: raise CustomerAdminError("committed activation receipt is missing or stale") @@ -1965,7 +2236,6 @@ receipt_registry != registry_path.resolve() or receipt_data != expected_data or receipt_audit != _activation_audit_path(profile_root).resolve() - or _document_fingerprint(document) != registry_sha256 or not expected_data.exists() or not expected_data.is_dir() or _path_has_symlink(expected_data, profile_root) @@ -1976,12 +2246,63 @@ ) ): raise CustomerAdminError("committed activation receipt is missing or stale") + if version == _CUSTOMER_ACTIVATION_RECEIPT_VERSION: + expected_keys = { + "version", + "source_journal_version", + "state", + "recovery_required", + "transaction_id", + "customer_id", + "registry_path", + "data_root", + "checklist_evidence_path", + "audit_path", + "audit_record_sha256", + "customer_projection_sha256", + "owner_sha256", + "admission_policy", + "admission_policy_sha256", + "created_at", + "committed_at", + } + if source_version == 1: + expected_keys.add("legacy_activation_receipt_sha256") + if source_version in {2, 3}: + expected_keys.add("nutrition_activation_receipt") + if source_version == 3: + expected_keys.update({ + "staff_membership_evidence_path", + "staff_membership_evidence_sha256", + "staff_chat_inventory_sha256", + "membership_subscription_epoch_id", + }) + policy = document.admission_policy + policy_payload = ( + policy.model_dump(mode="json", by_alias=True) + if policy is not None + else None + ) + if ( + set(receipt) != expected_keys + or not _audit_receipt_binds_customer_payload(receipt_audit, receipt) + or policy_payload is None + or receipt.get("customer_projection_sha256") + != _json_digest(spec.model_dump(mode="json")) + or receipt.get("owner_sha256") + != _json_digest(document.owner.model_dump(mode="json")) + or receipt.get("admission_policy") != policy_payload + or receipt.get("admission_policy_sha256") != _json_digest(policy_payload) + ): + raise CustomerAdminError("committed activation receipt is missing or stale") + elif _document_fingerprint(document) != registry_sha256: + raise CustomerAdminError("committed activation receipt is missing or stale") for key in ("created_at", "committed_at"): - value = journal.get(key) + value = receipt.get(key) if not isinstance(value, str): raise CustomerAdminError("committed activation receipt is missing or stale") try: - datetime.fromisoformat(value) + _ = datetime.fromisoformat(value) except ValueError as exc: raise CustomerAdminError( "committed activation receipt is missing or stale" @@ -1989,6 +2310,105 @@ return transaction_id +def _build_customer_activation_receipt( + journal: Mapping[str, JsonValue], + document: RegistryDocument, + spec: CustomerSpec, +) -> dict[str, JsonValue]: + policy = document.admission_policy + source_version = journal.get("version") + if policy is None or source_version not in {1, 2, 3}: + raise CustomerAdminError("committed activation receipt is missing or stale") + committed_at = journal.get("committed_at", journal.get("receipt_committed_at")) + receipt: dict[str, JsonValue] = { + "version": _CUSTOMER_ACTIVATION_RECEIPT_VERSION, + "source_journal_version": source_version, + "state": "committed", + "recovery_required": False, + "transaction_id": journal.get("transaction_id"), + "customer_id": spec.customer_key, + "registry_path": journal.get("registry_path"), + "data_root": journal.get("data_root"), + "checklist_evidence_path": journal.get("checklist_evidence_path"), + "audit_path": journal.get("audit_path"), + "audit_record_sha256": journal.get("audit_record_sha256"), + "customer_projection_sha256": _json_digest(spec.model_dump(mode="json")), + "owner_sha256": _json_digest(document.owner.model_dump(mode="json")), + "admission_policy": policy.model_dump(mode="json", by_alias=True), + "admission_policy_sha256": _json_digest( + policy.model_dump(mode="json", by_alias=True) + ), + "created_at": journal.get("created_at"), + "committed_at": committed_at, + } + if source_version == 1: + receipt["legacy_activation_receipt_sha256"] = _json_digest(journal) + if source_version in {2, 3}: + receipt["nutrition_activation_receipt"] = journal.get( + "nutrition_activation_receipt" + ) + if source_version == 3: + for key in ( + "staff_membership_evidence_path", + "staff_membership_evidence_sha256", + "staff_chat_inventory_sha256", + "membership_subscription_epoch_id", + ): + receipt[key] = journal.get(key) + return receipt + + +def _materialize_customer_activation_receipt( + profile_root: Path, + registry_path: Path, + document: RegistryDocument, + spec: CustomerSpec, + journal: Mapping[str, JsonValue], +) -> dict[str, JsonValue]: + path = _activation_receipt_path(profile_root, spec.customer_key) + existing = _read_activation_journal(path) + if existing is not None: + _ = _validate_committed_activation_payload( + profile_root, registry_path, document, spec, existing + ) + return existing + _ = _validate_committed_activation_payload( + profile_root, registry_path, document, spec, journal + ) + receipt = _build_customer_activation_receipt(journal, document, spec) + _write_activation_journal(path, receipt) + _ = _validate_committed_activation_payload( + profile_root, registry_path, document, spec, receipt + ) + return receipt + + +def _require_committed_activation_receipt( + profile_root: Path, + registry_path: Path, + document: RegistryDocument, + spec: CustomerSpec, +) -> str: + if document.admission_policy is None: + journal = _read_activation_journal(_activation_journal_path(profile_root)) + if journal is None: + raise CustomerAdminError("committed activation receipt is missing or stale") + return _validate_committed_activation_payload( + profile_root, registry_path, document, spec, journal + ) + receipt_path = _activation_receipt_path(profile_root, spec.customer_key) + receipt = _read_activation_journal(receipt_path) + if receipt is None: + journal = _read_activation_journal(_activation_journal_path(profile_root)) + if journal is None or journal.get("customer_id") != spec.customer_key: + raise CustomerAdminError("committed activation receipt is missing or stale") + receipt = _materialize_customer_activation_receipt( + profile_root, registry_path, document, spec, journal + ) + return _validate_committed_activation_payload( + profile_root, registry_path, document, spec, receipt + ) + def _validate_enabled_activation_receipts( profile_root: Path, registry_path: Path, @@ -6489,6 +6909,17 @@ return checked.date(), checked.isoformat(timespec="seconds") +def _gate_d_enabled_customer_isolated( + document: RegistryDocument, + customer_key: str, +) -> bool: + """Confirm one selected enabled customer within a valid bounded registry.""" + return any( + item.enabled and item.customer_key == customer_key + for item in document.customers + ) + + def audit_gate_d_preflight( profile_root: Path, customer_key: str, @@ -6575,7 +7006,7 @@ raise CustomerAdminError(f"unknown customer: {key}") if not selected.enabled: reasons.add("customer_not_enabled") - if len(enabled) != 1 or enabled[0].customer_key != key: + if not _gate_d_enabled_customer_isolated(document, key): reasons.add("main_profile_containment") else: checks["enabled_isolation"] = True diff -ruN '--exclude=.venv' '--exclude=__pycache__' '--exclude=.pytest_cache' '--exclude=.ruff_cache' '--exclude=t*' '--exclude=pc*' /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/customer_coaching.py /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/customer_coaching.py --- /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/customer_coaching.py 2026-08-25 22:21:00.860642402 +0900 +++ /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/customer_coaching.py 2026-08-27 14:43:51.559318368 +0900 @@ -5,11 +5,11 @@ import json import os +from collections.abc import Mapping from dataclasses import dataclass, field from datetime import date, time from pathlib import Path -from collections.abc import Mapping -from typing import Annotated, Literal +from typing import Annotated, ClassVar, Literal from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -982,6 +982,20 @@ raise CustomerRegistryError("customer inline plan must contain exactly 12 weeks") +class RegistryAdmissionPolicy(BaseModel): + """Explicit bounded authority for enabling multiple customers.""" + + model_config: ClassVar[ConfigDict] = ConfigDict( + extra="forbid", + frozen=True, + serialize_by_alias=True, + ) + + schema_name: Literal["nutricoach-multi-customer-v1"] = Field(alias="schema") + candidate_digest: str = Field(pattern=r"^[0-9a-f]{64}$") + max_enabled_customers: int = Field(ge=2, le=20) + + class RegistryDocument(BaseModel): model_config = ConfigDict(frozen=True) @@ -990,6 +1004,7 @@ diagnostic_session_digest: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$") owner: TelegramAddress customers: tuple[CustomerSpec, ...] + admission_policy: RegistryAdmissionPolicy | None = None @model_validator(mode="after") def require_unique_boundaries(self) -> RegistryDocument: @@ -998,6 +1013,10 @@ raise CustomerRegistryError( "diagnostic registry requires a diagnostic session binding" ) + if self.admission_policy is not None: + raise CustomerRegistryError( + "diagnostic registry cannot enable multi-customer admission" + ) elif self.diagnostic_session_digest is not None: raise CustomerRegistryError( "ordinary registry cannot carry a diagnostic session binding" @@ -1012,8 +1031,13 @@ if len(set(spaces)) != len(spaces): raise CustomerRegistryError("customer Telegram spaces must be unique") enabled = tuple(item for item in self.customers if item.enabled) - if len(enabled) > 1: - raise CustomerRegistryError("pilot supports only one external customer") + if len(enabled) > self.max_enabled_customers: + message = ( + "pilot supports only one external customer" + if self.admission_policy is None + else "enabled customer capacity exceeded" + ) + raise CustomerRegistryError(message) for customer in enabled: self.validate_enabled_customer(customer) return self @@ -1022,14 +1046,33 @@ self, candidate: CustomerSpec | None = None, ) -> None: - """Enforce the one-external-customer pilot boundary. + """Enforce the configured bounded customer capacity. Disabled entries remain parseable for historical/test data and do not - count toward the one-enabled-customer limit. + count toward the enabled-customer limit. """ - customers = self.customers + ((candidate,) if candidate is not None else ()) - if sum(item.enabled for item in customers) > 1: - raise CustomerRegistryError("pilot supports only one external customer") + if candidate is None: + customers = self.customers + else: + customers = tuple( + item + for item in self.customers + if item.customer_key != candidate.customer_key + ) + (candidate,) + if sum(item.enabled for item in customers) > self.max_enabled_customers: + message = ( + "pilot supports only one external customer" + if self.admission_policy is None + else "enabled customer capacity exceeded" + ) + raise CustomerRegistryError(message) + + @property + def max_enabled_customers(self) -> int: + """Return one unless the bounded capability is explicitly configured.""" + if self.admission_policy is None: + return 1 + return self.admission_policy.max_enabled_customers def validate_enabled_customer(self, spec: CustomerSpec) -> None: """Validate owner-inclusive identity and local activation requirements.""" @@ -1083,6 +1126,7 @@ class CustomerRegistry: owner: TelegramAddress customers: tuple[CustomerRuntime, ...] + admission_policy: RegistryAdmissionPolicy | None = None def load_customer_registry(path: Path, profile_root: Path) -> CustomerRegistry: @@ -1189,7 +1233,11 @@ mode=document.registry_mode, ) runtimes.append(CustomerRuntime(spec, data_root, binding)) - return CustomerRegistry(document.owner, tuple(runtimes)) + return CustomerRegistry( + document.owner, + tuple(runtimes), + document.admission_policy, + ) def _has_symlink(path: Path, profile_root: Path) -> bool: diff -ruN '--exclude=.venv' '--exclude=__pycache__' '--exclude=.pytest_cache' '--exclude=.ruff_cache' '--exclude=t*' '--exclude=pc*' /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/multi_customer_admission_migration.py /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/multi_customer_admission_migration.py --- /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/multi_customer_admission_migration.py 1970-01-01 09:00:00.000000000 +0900 +++ /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/multi_customer_admission_migration.py 2026-08-27 13:22:14.479070364 +0900 @@ -0,0 +1,112 @@ +"""Pure bounded multi-customer admission migration core.""" + +from __future__ import annotations + +import hashlib +import json +import re + +from checkin_cli.customer_coaching import ( + RegistryAdmissionPolicy, + RegistryDocument, +) +from checkin_cli.multi_customer_admission_migration_models import ( + AdmissionMigrationError, + AdmissionMigrationProposal, + AdmissionMigrationRejectReason, +) + + +def _sha256(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def _proposal_digest( + before_sha256: str, + after_sha256: str, + candidate_digest: str, + max_enabled_customers: int, +) -> str: + payload = json.dumps( + { + "after_sha256": after_sha256, + "before_sha256": before_sha256, + "capability": "nutricoach_multi_customer_v1", + "candidate_digest": candidate_digest, + "max_enabled_customers": max_enabled_customers, + }, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + return _sha256(payload) + + +def _parse_registry(payload: bytes) -> RegistryDocument: + try: + return RegistryDocument.model_validate_json(payload) + except (TypeError, ValueError) as exc: + raise AdmissionMigrationError(AdmissionMigrationRejectReason.REGISTRY) from exc + + +def propose_admission_migration( + before: bytes, + *, + candidate_digest: str, + max_enabled_customers: int, +) -> tuple[bytes, AdmissionMigrationProposal]: + """Build a deterministic, non-persisting capacity transition.""" + if re.fullmatch(r"[0-9a-f]{64}", candidate_digest) is None: + raise AdmissionMigrationError(AdmissionMigrationRejectReason.CANDIDATE) + document = _parse_registry(before) + if document.registry_mode != "ordinary_v1": + raise AdmissionMigrationError(AdmissionMigrationRejectReason.MODE) + if document.admission_policy is not None: + raise AdmissionMigrationError(AdmissionMigrationRejectReason.STATE) + try: + policy = RegistryAdmissionPolicy( + schema="nutricoach-multi-customer-v1", + candidate_digest=candidate_digest, + max_enabled_customers=max_enabled_customers, + ) + migrated = document.model_copy(update={"admission_policy": policy}) + migrated = RegistryDocument.model_validate(migrated.model_dump(mode="json")) + except (TypeError, ValueError) as exc: + raise AdmissionMigrationError(AdmissionMigrationRejectReason.CAPACITY) from exc + after = (migrated.model_dump_json(indent=2) + "\n").encode("utf-8") + before_sha256 = _sha256(before) + after_sha256 = _sha256(after) + proposal = AdmissionMigrationProposal( + before_sha256=before_sha256, + after_sha256=after_sha256, + candidate_digest=candidate_digest, + max_enabled_customers=max_enabled_customers, + proposal_digest=_proposal_digest( + before_sha256, + after_sha256, + candidate_digest, + max_enabled_customers, + ), + ) + return after, proposal + + +def apply_admission_migration( + before: bytes, + proposal: AdmissionMigrationProposal, + approval_phrase: str, +) -> bytes: + """Return exact after-bytes only for the sealed proposal and approval.""" + if _sha256(before) != proposal.before_sha256: + raise AdmissionMigrationError(AdmissionMigrationRejectReason.BEFORE) + after, expected = propose_admission_migration( + before, + candidate_digest=proposal.candidate_digest, + max_enabled_customers=proposal.max_enabled_customers, + ) + if expected != proposal: + raise AdmissionMigrationError(AdmissionMigrationRejectReason.PROPOSAL) + if approval_phrase != proposal.approval_phrase: + raise AdmissionMigrationError(AdmissionMigrationRejectReason.APPROVAL) + if _sha256(after) != proposal.after_sha256: + raise AdmissionMigrationError(AdmissionMigrationRejectReason.AFTER) + return after diff -ruN '--exclude=.venv' '--exclude=__pycache__' '--exclude=.pytest_cache' '--exclude=.ruff_cache' '--exclude=t*' '--exclude=pc*' /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/multi_customer_admission_migration_models.py /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/multi_customer_admission_migration_models.py --- /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/dualcoach/profile/checkin_cli/multi_customer_admission_migration_models.py 1970-01-01 09:00:00.000000000 +0900 +++ /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/dualcoach/profile/checkin_cli/multi_customer_admission_migration_models.py 2026-08-27 13:22:14.481070382 +0900 @@ -0,0 +1,48 @@ +"""Typed authority for bounded multi-customer admission migration.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from typing import override + + +class AdmissionMigrationRejectReason(StrEnum): + """Stable fail-closed migration reasons.""" + + AFTER = "after" + APPROVAL = "approval" + BEFORE = "before" + CANDIDATE = "candidate" + CAPACITY = "capacity" + MODE = "mode" + PROPOSAL = "proposal" + REGISTRY = "registry" + STATE = "state" + + +@dataclass(frozen=True, slots=True) +class AdmissionMigrationError(ValueError): + """A bounded admission migration was rejected before persistence.""" + + reason: AdmissionMigrationRejectReason + + @override + def __str__(self) -> str: + return self.reason.value + + +@dataclass(frozen=True, slots=True) +class AdmissionMigrationProposal: + """Immutable registry transition requiring exact owner approval.""" + + before_sha256: str + after_sha256: str + candidate_digest: str + max_enabled_customers: int + proposal_digest: str + + @property + def approval_phrase(self) -> str: + """Return the exact one-use approval phrase.""" + return f"AUTHORIZE NUTRICOACH MULTI CUSTOMER V1 {self.proposal_digest}" diff -ruN '--exclude=.venv' '--exclude=__pycache__' '--exclude=.pytest_cache' '--exclude=.ruff_cache' '--exclude=t*' '--exclude=pc*' /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/gateway/platforms/nutrition_coaching.py /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/gateway/platforms/nutrition_coaching.py --- /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/baseline/gateway/platforms/nutrition_coaching.py 2026-08-26 00:38:25.228186478 +0900 +++ /home/cube/projects/richard/.senpi-sandboxes/st_01a0405c-multi-customer-final-v17/ported/gateway/platforms/nutrition_coaching.py 2026-08-27 14:44:57.410842070 +0900 @@ -13,14 +13,14 @@ import sys import fcntl from contextlib import contextmanager, nullcontext -from collections.abc import Iterable, Iterator, Mapping, Sequence +from collections.abc import Awaitable, Iterable, Iterator, Mapping, Sequence from dataclasses import asdict, dataclass, replace from datetime import date, datetime, timedelta, timezone from decimal import Decimal from enum import StrEnum from pathlib import Path from threading import RLock -from typing import TYPE_CHECKING, Any, Awaitable, Callable, Literal, TypeGuard, cast +from typing import TYPE_CHECKING, Any, Callable, Literal, Protocol, TypeGuard, cast from zoneinfo import ZoneInfo logger = logging.getLogger(__name__) @@ -2198,6 +2198,16 @@ return registry, _resolve_committed_registry_path(root) +class _AdmissionPolicyView(Protocol): + @property + def candidate_digest(self) -> str: ... + + +class _AdmissionRegistryView(Protocol): + @property + def admission_policy(self) -> _AdmissionPolicyView | None: ... + + class NutritionCoachingCoordinator(WeeklyCoordinatorHost): """Route customer submissions and owner draft requests without generic ingress.""" @@ -2233,6 +2243,7 @@ self._weekly_host_authority: WeeklyHostAuthority = WeeklyHostAuthority.from_input( weekly_reminder_owner_input ) + self._require_multi_customer_candidate_authority(registry) if task26_runtime_required and ( task26_authority_source is None or task26_candidate_digest is None @@ -2289,9 +2300,23 @@ int(getattr(stat, "st_ctime_ns", 0)), ) + def _require_multi_customer_candidate_authority( + self, + registry: _AdmissionRegistryView, + ) -> None: + policy = registry.admission_policy + if ( + policy is not None + and self._task26_candidate_digest != policy.candidate_digest + ): + raise NutritionCoachingIntegrationError( + "multi-customer candidate authority mismatch" + ) + def _configure_registry(self, registry: CustomerRegistry) -> None: from checkin_cli.wizard import WizardService + self._require_multi_customer_candidate_authority(registry) self._registry = registry self._registered_routes = { tuple(str(value) for value in customer.spec.telegram.key): customer @@ -21547,6 +21572,14 @@ return receipt +class _ChannelInboxMetadataProvider(Protocol): + def channel_inbox_delivery_kwargs( + self, + chat_id: int | str, + topic_id: int | str, + ) -> dict[str, int | None]: ... + + class TelegramCustomerTransport: """Receipt-returning transport backed by one live Telegram adapter.""" @@ -21616,14 +21649,20 @@ ) if not callable(strict_sender): raise RuntimeError("Telegram strict topic sender is unavailable") - thread_kwargs = self._adapter._thread_kwargs_for_send( + channel_inbox_adapter = cast( + _ChannelInboxMetadataProvider, + self._adapter, + ) + thread_kwargs = channel_inbox_adapter.channel_inbox_delivery_kwargs( chat_id, topic_id, - {"thread_id": topic_id}, ) - if thread_kwargs.get("message_thread_id") is None: - raise RuntimeError( - "Telegram customer delivery requires message_thread_id" + if ( + thread_kwargs.get("message_thread_id") is None + and thread_kwargs.get("direct_messages_topic_id") is None + ): + raise NutritionCoachingIntegrationError( + "Telegram customer delivery requires an exact topic" ) return strict_sender, { "chat_id": chat_id, @@ -21906,7 +21945,7 @@ "adaptive delivery reservation is invalid" ) from exc except OSError as exc: - raise RuntimeError( + raise NutritionCoachingIntegrationError( "adaptive delivery consume outcome is unknown" ) from exc result = strict_sender(**send_kwargs) @@ -21943,7 +21982,7 @@ return {"ok": True, "message_id": receipt} -def _enabled_customer_key( +def select_enabled_customer_key( coordinator: NutritionCoachingCoordinator, requested: str | None, ) -> str: @@ -21989,7 +22028,7 @@ raise NutritionCoachingIntegrationError( "production console requires a receipt-returning customer transport" ) - selected_customer = _enabled_customer_key(coordinator, customer_key) + selected_customer = select_enabled_customer_key(coordinator, customer_key) try: owner = coordinator.owner except (AttributeError, TypeError, ValueError) as exc: