Deploy direct private KB runtime

README.md

@@ -3,6 +3,7 @@ title: Astera Customer AI
 sdk: docker
 app_port: 7860
 ---
+
 # Astera Customer AI
 
 Implementation repository for the current Customer AI runtime defined in the Astera Notion canon.

app.py

@@ -66,11 +66,26 @@ async def lifespan(_: FastAPI):
 
 
 app = FastAPI(title="Astera Customer AI", version="0.0.0", lifespan=lifespan)
-_origins = [
-    origin.strip()
-    for origin in os.environ.get("CUSTOMER_AI_ALLOWED_ORIGINS", "https://asterav8.jp").split(",")
-    if origin.strip()
-]
+_DEFAULT_ALLOWED_ORIGINS = (
+    "https://asterav8.jp",
+    "https://staging.asterav8.jp",
+    "https://open.asterav8.jp",
+    "https://localhost",
+    "capacitor://localhost",
+)
+
+
+def _merge_allowed_origins(configured: str) -> list[str]:
+    extra_origins = [
+        origin.strip()
+        for origin in configured.split(",")
+        if origin.strip()
+    ]
+    return list(dict.fromkeys([*_DEFAULT_ALLOWED_ORIGINS, *extra_origins]))
+
+
+_configured_origins = os.environ.get("CUSTOMER_AI_ALLOWED_ORIGINS", "").strip()
+_origins = _merge_allowed_origins(_configured_origins)
 app.add_middleware(
     CORSMiddleware,
     allow_origins=_origins,

config/quality.yaml

@@ -1,5 +1,5 @@
 quality:
-  user_need_resolution_min: 0.98
+  completion_primary_metric: answer_satisfaction
   answer_satisfaction_min: 0.98
   satisfaction_confidence_lower_bound_min: 0.98
   evaluation_min_unseen_scenarios: 200
@@ -9,9 +9,33 @@ quality:
   evaluation_min_multiturn: 20
   evaluation_min_false_premise: 20
   production_model_self_judge_allowed: false
-  critical_resolution_min: 0.99
+  user_need_resolution_role: diagnostic_only
+  critical_satisfaction_min: 0.99
   false_premise_correction_min: 1.0
   unsupported_hallucination_max: 0
   legacy_mixing_max: 0
   secret_leak_max: 0
   unexecuted_completion_claim_max: 0
+  satisfaction_required_dimensions:
+    - purpose_fulfilled
+    - preflight_correct
+    - intent_correct
+    - all_major_needs_covered
+    - required_depth_met
+    - factual
+    - evidence_complete
+    - constraints_respected
+    - conditions_exceptions_covered
+    - current_status_covered_when_required
+    - next_action_covered_when_required
+    - relevant
+    - direct
+    - clear
+    - appropriately_concise
+    - actionable_when_required
+    - context_consistent
+    - clarification_efficient
+    - resolution_mode_correct
+    - self_contained
+  same_revision_evidence_required: true
+  evaluation_input_contract: scenario_plus_runtime_evidence

deployment-manifest.json

@@ -0,0 +1,7 @@
+{
+  "schema_version": 5,
+  "kb_storage": "huggingface_private_bucket_direct_runtime",
+  "kb_bucket_id": "G-ACE/astera-customerai-kb",
+  "kb_build_id": "kb-20260814T042741+0900",
+  "kb_embedded_in_space": false
+}

pyproject.toml

@@ -2,11 +2,11 @@
 name = "astera-customer-ai"
 version = "0.0.0"
 requires-python = ">=3.11"
-dependencies = ["fastapi>=0.128,<1", "uvicorn>=0.48,<1", "pydantic>=2.12,<3", "httpx>=0.28,<1", "rapidfuzz==3.14.3", "pyyaml>=6,<7"]
+dependencies = ["fastapi>=0.128,<1", "uvicorn>=0.48,<1", "pydantic>=2.12,<3", "httpx>=0.28,<1", "rapidfuzz==3.14.3", "pyyaml>=6,<7", "huggingface_hub>=1.24,<2"]
 
 [project.optional-dependencies]
 training = ["datasets", "transformers", "peft", "trl"]
-test = ["pytest==9.0.2", "pytest-asyncio>=1,<2", "ruff>=0.12,<1"]
+test = ["pytest==9.0.2", "pytest-asyncio>=1,<2", "ruff>=0.12,<1", "huggingface_hub>=1.24,<2"]
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"

requirements.txt

@@ -4,3 +4,4 @@ pydantic>=2.12,<3
 httpx>=0.28,<1
 rapidfuzz==3.14.3
 pyyaml>=6,<7
+huggingface_hub>=1.24,<2

runtime/internal_core.py

@@ -86,9 +86,11 @@ class CustomerAIInternalCore:
         }
         contract = None
         try:
-            contract = self.decomposer.decompose(normalized_text, context)
-            if self.decomposer.requires_semantic_expansion(contract) and hasattr(self.roles, "semantic_decompose"):
-                contract = await self.roles.semantic_decompose(normalized_text, contract)
+            seed_contract = self.decomposer.decompose(normalized_text, context)
+            contract = seed_contract
+            if self.decomposer.requires_semantic_expansion(seed_contract) and hasattr(self.roles, "semantic_decompose"):
+                semantic_contract = await self.roles.semantic_decompose(normalized_text, seed_contract)
+                contract = self.decomposer.protect_semantic_expansion(seed_contract, semantic_contract)
             bound_tasks = self.state.bind_tasks(session_id, contract.need_tasks, follow_up_kind)
             contract = contract.model_copy(update={"need_tasks": bound_tasks})
             state = self.state.get(session_id)
@@ -104,25 +106,43 @@ class CustomerAIInternalCore:
                 user_conditions=dict(state.user_conditions),
             )
         except GroundingConflictError:
+            unresolved = [t.task_id for t in contract.need_tasks] if contract is not None else []
             if contract is not None:
                 self.state.complete_turn(
                     session_id,
                     contract.need_tasks,
                     resolved_task_ids=set(),
-                    unresolved_task_ids={t.task_id for t in contract.need_tasks},
+                    unresolved_task_ids=set(unresolved),
                     satisfaction_blockers={"grounding_conflict"},
                 )
-            return self._failure(request_id, session_id, turn_id, ResolutionMode.BLOCKED_CURRENT_FACT, "grounding_conflict", ["grounding_conflict"])
+            return self._failure(
+                request_id,
+                session_id,
+                turn_id,
+                ResolutionMode.BLOCKED_CURRENT_FACT,
+                "grounding_conflict",
+                ["grounding_conflict"],
+                unresolved_task_ids=unresolved,
+            )
         except Exception:
+            unresolved = [t.task_id for t in contract.need_tasks] if contract is not None else []
             if contract is not None:
                 self.state.complete_turn(
                     session_id,
                     contract.need_tasks,
                     resolved_task_ids=set(),
-                    unresolved_task_ids={t.task_id for t in contract.need_tasks},
+                    unresolved_task_ids=set(unresolved),
                     satisfaction_blockers={"preflight_runtime_failure"},
                 )
-            return self._failure(request_id, session_id, turn_id, ResolutionMode.RUNTIME_FAILURE, "runtime_failure", ["preflight_runtime_failure"])
+            return self._failure(
+                request_id,
+                session_id,
+                turn_id,
+                ResolutionMode.RUNTIME_FAILURE,
+                "runtime_failure",
+                ["preflight_runtime_failure"],
+                unresolved_task_ids=unresolved,
+            )
 
         language = "ja" if any("\u3040" <= ch <= "\u30ff" or "\u4e00" <= ch <= "\u9fff" for ch in normalized_text) else "en"
         audience = self._audience(normalized_text)
@@ -159,14 +179,23 @@ class CustomerAIInternalCore:
             results = await self.roles.run_all(packet, capsules)
             integrated = self.integrator.integrate(results)
         except Exception:
+            unresolved = [t.task_id for t in contract.need_tasks]
             self.state.complete_turn(
                 session_id,
                 contract.need_tasks,
                 resolved_task_ids=set(),
-                unresolved_task_ids={t.task_id for t in contract.need_tasks},
+                unresolved_task_ids=set(unresolved),
                 satisfaction_blockers={"role_runtime_failure"},
             )
-            return self._failure(request_id, session_id, turn_id, ResolutionMode.RUNTIME_FAILURE, "runtime_failure", ["role_runtime_failure"])
+            return self._failure(
+                request_id,
+                session_id,
+                turn_id,
+                ResolutionMode.RUNTIME_FAILURE,
+                "runtime_failure",
+                ["role_runtime_failure"],
+                unresolved_task_ids=unresolved,
+            )
 
         external = self.audit.check(packet, results)
         quality = self.gate.evaluate(packet, integrated, external_violations=external)
@@ -204,7 +233,12 @@ class CustomerAIInternalCore:
         composed = self.composer.compose(plan)
         answer = self.refiner.refine(composed.answer or "") if composed.answer else ""
         terminology = self.japanese.terminology_violations(answer) if answer else []
-        security = self.security.check_output(answer=answer, forbidden_literals=packet.forbidden_claims, unexecuted_completion_claim=False)
+        unexecuted_claim = self.security.detect_unexecuted_completion_claim(answer) if answer else False
+        security = self.security.check_output(
+            answer=answer,
+            forbidden_literals=packet.forbidden_claims,
+            unexecuted_completion_claim=unexecuted_claim,
+        )
         major = [t for t in contract.need_tasks if t.priority == "primary"]
         resolved = set(composed.resolved_task_ids)
         all_major = all(t.task_id in resolved for t in major)
@@ -241,10 +275,31 @@ class CustomerAIInternalCore:
         unresolved_task_ids = set(composed.unresolved_task_ids)
         if not passed:
             unresolved_task_ids.update(t.task_id for t in contract.need_tasks if t.task_id not in resolved)
+
+        zero_tolerance = {"unsupported_claim", "forbidden_literal_exposed", "unexecuted_completion_claim"}
+        public_blocked = bool(zero_tolerance.intersection(violations))
+        public_answer = answer or composed.answer
+        public_mode = composed.mode
+        answered_task_ids = set(composed.resolved_task_ids)
+        state_resolved = set(resolved)
+        if public_blocked:
+            public_answer = None
+            public_mode = ResolutionMode.SAFETY_BLOCKED
+            answered_task_ids.clear()
+            unresolved_task_ids.update(t.task_id for t in contract.need_tasks)
+            state_resolved.clear()
+        elif not passed and composed.mode == ResolutionMode.RESOLVED:
+            # A fully composed answer that failed a non-zero-tolerance quality gate
+            # must not be advertised or persisted as resolved. Keep the text as a
+            # safe partial response, but keep the need open for repair/follow-up.
+            public_mode = ResolutionMode.SAFE_PARTIAL
+            unresolved_task_ids.update(t.task_id for t in contract.need_tasks)
+            state_resolved.clear()
+
         self.state.complete_turn(
             session_id,
             contract.need_tasks,
-            resolved_task_ids=resolved,
+            resolved_task_ids=state_resolved,
             unresolved_task_ids=unresolved_task_ids,
             evidence_gaps=set(quality.missing_evidence_task_ids),
             satisfaction_blockers=set(violations),
@@ -254,37 +309,46 @@ class CustomerAIInternalCore:
         if not passed:
             if "grounding_conflict" in violations:
                 failure_class = "grounding_conflict"
+            elif public_blocked or not security.passed:
+                failure_class = "safety_rejection"
             elif {"major_need_missing", "evidence_incomplete", "conversation_not_resolved"}.intersection(violations):
                 failure_class = "coverage_defect"
-            elif not security.passed:
-                failure_class = "safety_rejection"
             else:
                 failure_class = "runtime_failure"
         return FinalResponse(
             request_id=request_id,
             session_id=session_id,
             turn_id=turn_id,
-            answer=answer or composed.answer,
-            answered_task_ids=list(composed.resolved_task_ids),
-            unresolved_task_ids=list(composed.unresolved_task_ids),
+            answer=public_answer,
+            answered_task_ids=sorted(answered_task_ids),
+            unresolved_task_ids=sorted(unresolved_task_ids),
             evidence_ids=list(integrated.evidence_ids),
             resolution_score=quality.resolution_score,
             passed=passed,
-            resolution_mode=composed.mode,
+            resolution_mode=public_mode,
             clarification_questions=list(composed.clarification_questions),
             failure_class=failure_class,
             violations=violations,
         )
 
     @staticmethod
-    def _failure(request_id, session_id, turn_id, mode, failure_class, violations):
+    def _failure(
+        request_id,
+        session_id,
+        turn_id,
+        mode,
+        failure_class,
+        violations,
+        *,
+        unresolved_task_ids=None,
+    ):
         return FinalResponse(
             request_id=request_id,
             session_id=session_id,
             turn_id=turn_id,
             answer=None,
             answered_task_ids=[],
-            unresolved_task_ids=[],
+            unresolved_task_ids=list(unresolved_task_ids or ()),
             evidence_ids=[],
             resolution_score=0.0,
             passed=False,

runtime/kb_bucket.py

@@ -0,0 +1,216 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+
+from huggingface_hub import HfFileSystem
+
+HF_KB_BUCKET_DEFAULT = "G-ACE/astera-customerai-kb"
+HF_KB_MOUNT_DEFAULT = "/data/customer-ai"
+HF_KB_ACTIVE_POINTER_DEFAULT = "active.json"
+
+
+@dataclass(frozen=True)
+class MountedKBRelease:
+    build_id: str
+    canonical_path: Path
+    current_facts_path: Path | None
+    aliases_path: Path | None
+
+
+def _resolve_file_inside_mount(mount: Path, relative_path: str, code: str) -> Path:
+    relative = relative_path.strip()
+    if not relative:
+        raise ValueError(code)
+    mount_resolved = mount.resolve()
+    candidate = (mount_resolved / relative).resolve()
+    if candidate != mount_resolved and mount_resolved not in candidate.parents:
+        raise ValueError("kb_pointer_path_escape")
+    if not candidate.is_file() or candidate.stat().st_size == 0:
+        raise ValueError(code)
+    return candidate
+
+
+def _safe_relative(value: str, code: str) -> str:
+    relative = value.strip()
+    if not relative:
+        raise ValueError(code)
+    path = PurePosixPath(relative)
+    if path.is_absolute() or ".." in path.parts:
+        raise ValueError("kb_pointer_path_escape")
+    return path.as_posix()
+
+
+def _validate_pointer(payload: object, expected_build_id: str) -> tuple[str, str, str | None, str | None, str]:
+    if not isinstance(payload, dict):
+        raise ValueError("kb_active_pointer_invalid")
+
+    build_id = str(payload.get("build_id") or "").strip()
+    expected = expected_build_id.strip()
+    if not build_id:
+        raise ValueError("kb_active_build_id_missing")
+    if not expected:
+        raise ValueError("kb_expected_build_id_missing")
+    if build_id != expected:
+        raise ValueError("kb_active_build_id_mismatch")
+
+    release_prefix = f"releases/{expected}/"
+    canonical = _safe_relative(str(payload.get("canonical_path") or ""), "kb_active_canonical_missing")
+    manifest = _safe_relative(str(payload.get("manifest_path") or ""), "kb_active_manifest_missing")
+    if not canonical.startswith(release_prefix) or not manifest.startswith(release_prefix):
+        raise ValueError("kb_pointer_release_path_mismatch")
+
+    current_raw = str(payload.get("current_facts_path") or "").strip()
+    current = _safe_relative(current_raw, "kb_active_current_missing") if current_raw else None
+    if current is not None and not current.startswith(release_prefix):
+        raise ValueError("kb_pointer_release_path_mismatch")
+
+    aliases_raw = str(payload.get("aliases_path") or "").strip()
+    aliases = _safe_relative(aliases_raw, "kb_active_aliases_missing") if aliases_raw else None
+    if aliases is not None and not aliases.startswith(release_prefix):
+        raise ValueError("kb_pointer_release_path_mismatch")
+
+    return build_id, canonical, current, aliases, manifest
+
+
+def load_mounted_kb_release(
+    *,
+    mount_path: str,
+    expected_build_id: str,
+    pointer_name: str = HF_KB_ACTIVE_POINTER_DEFAULT,
+) -> MountedKBRelease:
+    mount = Path(mount_path).expanduser()
+    if not mount.is_dir():
+        raise ValueError("kb_bucket_mount_missing")
+
+    pointer = _resolve_file_inside_mount(mount, pointer_name, "kb_active_pointer_missing")
+    try:
+        payload = json.loads(pointer.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise ValueError("kb_active_pointer_invalid") from exc
+
+    build_id, canonical_value, current_value, aliases_value, manifest_value = _validate_pointer(
+        payload,
+        expected_build_id,
+    )
+    _resolve_file_inside_mount(mount, manifest_value, "kb_active_manifest_missing")
+
+    canonical_path = _resolve_file_inside_mount(mount, canonical_value, "kb_active_canonical_missing")
+    current_facts_path = (
+        _resolve_file_inside_mount(mount, current_value, "kb_active_current_missing")
+        if current_value
+        else None
+    )
+    aliases_path = (
+        _resolve_file_inside_mount(mount, aliases_value, "kb_active_aliases_missing")
+        if aliases_value
+        else None
+    )
+
+    return MountedKBRelease(
+        build_id=build_id,
+        canonical_path=canonical_path,
+        current_facts_path=current_facts_path,
+        aliases_path=aliases_path,
+    )
+
+
+def load_remote_kb_release(
+    *,
+    bucket_id: str,
+    token: str,
+    expected_build_id: str,
+    pointer_name: str = HF_KB_ACTIVE_POINTER_DEFAULT,
+    cache_root: str | None = None,
+) -> MountedKBRelease:
+    bucket = bucket_id.strip()
+    auth = token.strip()
+    if not bucket:
+        raise ValueError("kb_bucket_id_missing")
+    if not auth:
+        raise ValueError("hf_token_missing")
+
+    fs = HfFileSystem(token=auth)
+    base = f"hf://buckets/{bucket}"
+
+    def read_remote(relative_path: str, code: str) -> bytes:
+        relative = _safe_relative(relative_path, code)
+        try:
+            with fs.open(f"{base}/{relative}", "rb") as handle:
+                raw = handle.read()
+        except Exception as exc:
+            raise ValueError(code) from exc
+        if not raw:
+            raise ValueError(code)
+        return raw
+
+    pointer_name = _safe_relative(pointer_name, "kb_active_pointer_missing")
+    try:
+        pointer_payload = json.loads(read_remote(pointer_name, "kb_active_pointer_missing").decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ValueError("kb_active_pointer_invalid") from exc
+
+    build_id, canonical_value, current_value, aliases_value, manifest_value = _validate_pointer(
+        pointer_payload,
+        expected_build_id,
+    )
+
+    try:
+        manifest_payload = json.loads(read_remote(manifest_value, "kb_active_manifest_missing").decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ValueError("kb_active_manifest_invalid") from exc
+    if not isinstance(manifest_payload, dict) or str(manifest_payload.get("build_id") or "") != build_id:
+        raise ValueError("kb_active_manifest_invalid")
+
+    manifest_files: dict[str, tuple[int, str]] = {}
+    for item in manifest_payload.get("files") or []:
+        if not isinstance(item, dict):
+            continue
+        name = str(item.get("path") or "").strip()
+        sha256 = str(item.get("sha256") or "").strip()
+        try:
+            size = int(item.get("bytes"))
+        except (TypeError, ValueError):
+            continue
+        if name and sha256:
+            manifest_files[name] = (size, sha256)
+
+    cache = Path(cache_root).expanduser() if cache_root else Path(tempfile.gettempdir()) / "astera-customerai-kb"
+    release_cache = cache / build_id
+    release_cache.mkdir(parents=True, exist_ok=True)
+
+    def materialize(relative_path: str, code: str) -> Path:
+        relative = _safe_relative(relative_path, code)
+        name = PurePosixPath(relative).name
+        expected = manifest_files.get(name)
+        if expected is None:
+            raise ValueError("kb_manifest_file_missing")
+        expected_size, expected_sha = expected
+        destination = release_cache / name
+
+        if destination.is_file():
+            cached = destination.read_bytes()
+            if len(cached) == expected_size and hashlib.sha256(cached).hexdigest() == expected_sha:
+                return destination
+
+        raw = read_remote(relative, code)
+        if len(raw) != expected_size or hashlib.sha256(raw).hexdigest() != expected_sha:
+            raise ValueError("kb_remote_integrity_mismatch")
+        temporary = destination.with_suffix(destination.suffix + ".tmp")
+        temporary.write_bytes(raw)
+        temporary.replace(destination)
+        return destination
+
+    canonical_path = materialize(canonical_value, "kb_active_canonical_missing")
+    current_facts_path = materialize(current_value, "kb_active_current_missing") if current_value else None
+    aliases_path = materialize(aliases_value, "kb_active_aliases_missing") if aliases_value else None
+
+    return MountedKBRelease(
+        build_id=build_id,
+        canonical_path=canonical_path,
+        current_facts_path=current_facts_path,
+        aliases_path=aliases_path,
+    )

runtime/security.py

@@ -1,11 +1,23 @@
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass
 from typing import Iterable
 
 from .schemas import GroundedFact
 
 
+_UNEXECUTED_JA = re.compile(
+    r"(?:^|[。！？!?\n])\s*(?:[-*・]\s*)?(?:(?:こちら|当方)で\s*|私(?:が|は)?\s*)?"
+    r"(?:(?:実行|変更|更新|削除|作成|送信|登録|設定|反映|デプロイ|公開|保存|修正|コミット|プッシュ|処理)(?:を)?(?:し|いたし)(?:ました|ておきました|てあります)|完了しました)"
+)
+_UNEXECUTED_EN = re.compile(
+    r"(?:^|[.!?\n])\s*(?:[-*]\s*)?(?:i|we)\s+(?:have\s+)?"
+    r"(?:executed|changed|updated|deleted|created|sent|registered|configured|deployed|published|saved|fixed|committed|pushed|completed)\b",
+    re.IGNORECASE,
+)
+
+
 @dataclass(frozen=True)
 class SecurityCheck:
     passed: bool
@@ -16,11 +28,31 @@ class PublicBoundary:
     def filter_facts(self, facts: Iterable[GroundedFact]) -> list[GroundedFact]:
         return [f for f in facts if f.public and not f.legacy and not f.undecided]
 
-    def check_output(self, *, answer: str, forbidden_literals: Iterable[str], unexecuted_completion_claim: bool) -> SecurityCheck:
+    @staticmethod
+    def detect_unexecuted_completion_claim(answer: str) -> bool:
+        """Detect assistant-style claims that an external mutation was already executed.
+
+        Customer AI has no write/deploy authority. A role prompt is not sufficient as
+        an enforcement boundary, so suspicious first-person/subject-omitted completion
+        language is rejected deterministically before public output.
+        """
+
+        if not answer.strip():
+            return False
+        return bool(_UNEXECUTED_JA.search(answer) or _UNEXECUTED_EN.search(answer))
+
+    def check_output(
+        self,
+        *,
+        answer: str,
+        forbidden_literals: Iterable[str],
+        unexecuted_completion_claim: bool,
+    ) -> SecurityCheck:
         violations: list[str] = []
         for literal in forbidden_literals:
             if literal and literal in answer:
-                violations.append("forbidden_literal_exposed"); break
-        if unexecuted_completion_claim:
+                violations.append("forbidden_literal_exposed")
+                break
+        if unexecuted_completion_claim or self.detect_unexecuted_completion_claim(answer):
             violations.append("unexecuted_completion_claim")
         return SecurityCheck(passed=not violations, violations=violations)

runtime/startup.py

@@ -7,6 +7,13 @@ from pathlib import Path
 
 from .bootstrap import RuntimeDependencies, build_work
 from .hf_client import HF_CHAT_API
+from .kb_bucket import (
+    HF_KB_ACTIVE_POINTER_DEFAULT,
+    HF_KB_BUCKET_DEFAULT,
+    HF_KB_MOUNT_DEFAULT,
+    load_mounted_kb_release,
+    load_remote_kb_release,
+)
 from .live_state import EmptyLiveStateProvider, HybridLiveStateProvider
 from .service import CustomerAIWork
 
@@ -24,6 +31,13 @@ def _required_file(value: str, code: str) -> Path:
     return path
 
 
+def _required_value(values: Mapping[str, str], key: str, code: str) -> str:
+    value = values.get(key, "").strip()
+    if not value:
+        raise RuntimeNotReady(code)
+    return value
+
+
 def _load_alias_registry(path_value: str) -> Mapping[str, Iterable[str]]:
     if not path_value.strip():
         return {}
@@ -42,18 +56,67 @@ def _load_alias_registry(path_value: str) -> Mapping[str, Iterable[str]]:
     return normalized
 
 
+def _production_kb_files(values: Mapping[str, str]) -> tuple[Path, Path | None, Path | None, str]:
+    build_id = _required_value(values, "CUSTOMER_AI_KB_BUILD_ID", "kb_build_id_missing")
+    mount_path = values.get("CUSTOMER_AI_KB_MOUNT_PATH", "").strip() or HF_KB_MOUNT_DEFAULT
+    pointer_name = (
+        values.get("CUSTOMER_AI_KB_ACTIVE_POINTER", "").strip()
+        or HF_KB_ACTIVE_POINTER_DEFAULT
+    )
+    bucket_id = values.get("CUSTOMER_AI_KB_BUCKET_ID", "").strip() or HF_KB_BUCKET_DEFAULT
+    token = (values.get("HF_TOKEN", "") or values.get("HF_KEY", "")).strip()
+
+    try:
+        release = load_mounted_kb_release(
+            mount_path=mount_path,
+            expected_build_id=build_id,
+            pointer_name=pointer_name,
+        )
+    except ValueError as exc:
+        if str(exc) != "kb_bucket_mount_missing":
+            raise RuntimeNotReady(str(exc)) from exc
+        if not token:
+            raise RuntimeNotReady("hf_token_missing") from exc
+        try:
+            release = load_remote_kb_release(
+                bucket_id=bucket_id,
+                token=token,
+                expected_build_id=build_id,
+                pointer_name=pointer_name,
+            )
+        except ValueError as remote_exc:
+            raise RuntimeNotReady(str(remote_exc)) from remote_exc
+
+    return (
+        release.canonical_path,
+        release.current_facts_path,
+        release.aliases_path,
+        release.build_id,
+    )
+
+
 def create_work_from_environment(
     env: Mapping[str, str] | None = None,
     *,
     role_pool: object | None = None,
 ) -> CustomerAIWork:
     values = os.environ if env is None else env
-    kb_path = _required_file(values.get("CUSTOMER_AI_KB_SNAPSHOT_PATH", "").strip(), "kb_snapshot_missing")
-    generation_id = values.get("CUSTOMER_AI_KB_GENERATION_ID", "").strip() or kb_path.stem
 
-    current_value = values.get("CUSTOMER_AI_CURRENT_FACTS_PATH", "").strip()
-    if current_value:
-        current_path = _required_file(current_value, "current_facts_missing")
+    if role_pool is None:
+        kb_path, current_path, alias_path, build_id = _production_kb_files(values)
+        generation_id = values.get("CUSTOMER_AI_KB_GENERATION_ID", "").strip() or build_id
+    else:
+        kb_path = _required_file(
+            values.get("CUSTOMER_AI_KB_SNAPSHOT_PATH", "").strip(),
+            "kb_snapshot_missing",
+        )
+        generation_id = values.get("CUSTOMER_AI_KB_GENERATION_ID", "").strip() or kb_path.stem
+        current_value = values.get("CUSTOMER_AI_CURRENT_FACTS_PATH", "").strip()
+        current_path = _required_file(current_value, "current_facts_missing") if current_value else None
+        alias_value = values.get("CUSTOMER_AI_ALIAS_REGISTRY_PATH", "").strip()
+        alias_path = _required_file(alias_value, "alias_registry_missing") if alias_value else None
+
+    if current_path is not None:
         live_provider = HybridLiveStateProvider.from_jsonl(
             current_path,
             generation_id=f"{generation_id}:current",
@@ -73,7 +136,7 @@ def create_work_from_environment(
     return build_work(
         RuntimeDependencies(
             live_state_provider=live_provider,
-            japanese_alias_registry=_load_alias_registry(values.get("CUSTOMER_AI_ALIAS_REGISTRY_PATH", "")),
+            japanese_alias_registry=_load_alias_registry(str(alias_path) if alias_path else ""),
             japanese_fuzzy_threshold=fuzzy_threshold,
             kb_snapshot_path=str(kb_path),
             kb_generation_id=generation_id,

runtime/task_decomposition.py

@@ -7,6 +7,8 @@ from .contracts import TaskContract
 from .schemas import NeedTask
 
 _SENTENCE_SPLIT = re.compile(r"(?:\n{2,}|(?<=[。！？!?])\s*)")
+_COVERAGE_SPLIT = re.compile(r"(?:\n+|[。！？!?、,]+|(?:さらに|加えて|また|なお|かつ|一方で))")
+_COVERAGE_NOISE = re.compile(r"[\s\u3000・:：;；（）()「」『』【】\[\]\"'`]+")
 _COMPARISON = ("比較", "違い", "どちら", "どっち", "vs", "対して")
 _PROCEDURE = ("方法", "手順", "やり方", "どうや", "設定", "登録", "作成", "実装")
 _TROUBLE = ("エラー", "不具合", "動か", "失敗", "直ら", "できない", "困")
@@ -41,6 +43,19 @@ class TaskDecomposer:
         folded = text.casefold()
         return not any(k in folded for k in ("ありがとう", "thanks", "thank you", "こんにちは", "hello"))
 
+    @staticmethod
+    def _coverage_normalize(text: str) -> str:
+        return _COVERAGE_NOISE.sub("", text.casefold())
+
+    @classmethod
+    def _coverage_fragments(cls, text: str) -> tuple[str, ...]:
+        fragments: list[str] = []
+        for part in _COVERAGE_SPLIT.split(text):
+            normalized = cls._coverage_normalize(part)
+            if len(normalized) >= 2:
+                fragments.append(normalized)
+        return tuple(dict.fromkeys(fragments))
+
     def _make_task(self, text: str, idx: int, *, priority: str = "primary") -> NeedTask:
         shape = self._shape(text)
         return NeedTask(
@@ -83,3 +98,30 @@ class TaskDecomposer:
 
     def requires_semantic_expansion(self, contract: TaskContract) -> bool:
         return "semantic_decomposition_required" in contract.constraints
+
+    def protect_semantic_expansion(self, seed: TaskContract, candidate: TaskContract) -> TaskContract:
+        """Accept semantic expansion only when preservation of the original need is provable.
+
+        The semantic model may paraphrase freely, but Customer AI must never silently
+        drop a user clause. If deterministic coverage cannot be proven, fall back to
+        the seed's full-request task rather than trusting the model decomposition.
+        """
+
+        if not candidate.need_tasks:
+            return seed
+        coverage_text = " ".join(task.text for task in candidate.need_tasks)
+        coverage = self._coverage_normalize(coverage_text)
+        required = self._coverage_fragments(seed.target)
+        if required and not all(fragment in coverage for fragment in required):
+            return seed.model_copy(
+                update={
+                    "constraints": [item for item in seed.constraints if item != "semantic_decomposition_required"]
+                }
+            )
+        return candidate.model_copy(
+            update={
+                "target": seed.target,
+                "conditions": list(dict.fromkeys([*seed.conditions, *candidate.conditions])),
+                "constraints": [item for item in candidate.constraints if item != "semantic_decomposition_required"],
+            }
+        )