feat: add intent/proposals v1 subjects and schemas

src/qiki/shared/models/orion_qiki_protocol.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, Literal, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+class _StrictModel(BaseModel):
+    model_config = ConfigDict(extra="forbid", validate_assignment=True)
+
+
+class LangHint(str, Enum):
+    AUTO = "auto"
+    RU = "ru"
+    EN = "en"
+
+
+class EnvironmentMode(str, Enum):
+    FACTORY = "FACTORY"
+    MISSION = "MISSION"
+
+
+class SelectionV1(_StrictModel):
+    kind: Literal["event", "incident", "track", "snapshot", "none"] = "none"
+    id: Optional[str] = None
+
+
+class IntentV1(_StrictModel):
+    version: Literal[1] = 1
+    text: str
+    lang_hint: LangHint = LangHint.AUTO
+    screen: str
+    selection: SelectionV1 = Field(default_factory=SelectionV1)
+    ts: int
+    environment_mode: EnvironmentMode = EnvironmentMode.FACTORY
+    snapshot_min: dict[str, Any] = Field(default_factory=dict)
+
+
+class ProposalV1(_StrictModel):
+    proposal_id: str
+    title: str
+    justification: str
+    priority: int = Field(ge=0, le=100)
+    confidence: float = Field(ge=0.0, le=1.0)
+    proposed_actions: list[Any] = Field(default_factory=list)
+
+    @field_validator("proposed_actions")
+    @classmethod
+    def _must_be_empty_in_stage_a(cls, v: list[Any]) -> list[Any]:
+        if v:
+            raise ValueError("proposed_actions must be empty in Stage A")
+        return v
+
+
+class ProposalsBatchV1(_StrictModel):
+    version: Literal[1] = 1
+    ts: int
+    proposals: list[ProposalV1] = Field(default_factory=list)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+

src/qiki/shared/nats_subjects.py

@@ -30,7 +30,11 @@
 RESPONSES_CONTROL = "qiki.responses.control"
 
 # QIKI interaction subjects (operator intents, agent replies)
-QIKI_INTENTS = "qiki.intents"
+QIKI_INTENT_V1 = "qiki.intent.v1"
+QIKI_PROPOSALS_V1 = "qiki.proposals.v1"
+
+# Backward-compat alias: prefer QIKI_INTENT_V1.
+QIKI_INTENTS = QIKI_INTENT_V1
 
 # Events subjects
 EVENTS_V1_WILDCARD = "qiki.events.v1.>"

tests/unit/test_orion_qiki_protocol_v1.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+from pydantic import ValidationError
+
+from qiki.shared.models.orion_qiki_protocol import (
+    EnvironmentMode,
+    IntentV1,
+    LangHint,
+    ProposalV1,
+    ProposalsBatchV1,
+    SelectionV1,
+)
+
+
+def test_intent_v1_roundtrip() -> None:
+    payload = IntentV1(
+        text="scan 360",
+        lang_hint=LangHint.EN,
+        screen="Events/События",
+        selection=SelectionV1(kind="incident", id="INC|sensor|core"),
+        ts=1700000000000,
+        environment_mode=EnvironmentMode.FACTORY,
+        snapshot_min={"nats": True, "unread": 3},
+    )
+    dumped = payload.model_dump()
+    reloaded = IntentV1.model_validate(dumped)
+    assert reloaded.version == 1
+    assert reloaded.text == "scan 360"
+    assert reloaded.selection.kind == "incident"
+    assert reloaded.snapshot_min["unread"] == 3
+
+
+def test_intent_v1_requires_fields() -> None:
+    with pytest.raises(ValidationError):
+        IntentV1.model_validate({"version": 1, "text": "x"})
+
+
+def test_proposal_v1_actions_must_be_empty_in_stage_a() -> None:
+    ok = ProposalV1(
+        proposal_id="P1",
+        title="Title",
+        justification="Justification",
+        priority=50,
+        confidence=0.6,
+        proposed_actions=[],
+    )
+    assert ok.proposed_actions == []
+
+    with pytest.raises(ValidationError):
+        ProposalV1(
+            proposal_id="P2",
+            title="Title",
+            justification="Justification",
+            priority=50,
+            confidence=0.6,
+            proposed_actions=[{"op": "do"}],
+        )
+
+
+def test_batch_v1_json_roundtrip() -> None:
+    batch = ProposalsBatchV1(
+        ts=1700000000000,
+        proposals=[
+            ProposalV1(
+                proposal_id="P1",
+                title="T",
+                justification="J",
+                priority=10,
+                confidence=0.9,
+            )
+        ],
+        metadata={"request_id": "RID"},
+    )
+    raw = batch.model_dump_json()
+    parsed = ProposalsBatchV1.model_validate_json(raw)
+    assert parsed.version == 1
+    assert parsed.proposals[0].proposal_id == "P1"
+    assert parsed.metadata["request_id"] == "RID"
+
+
+def test_version_compatibility_strict() -> None:
+    payload = {
+        "version": 2,
+        "ts": 1700000000000,
+        "proposals": [],
+        "metadata": {},
+    }
+    with pytest.raises(ValidationError):
+        ProposalsBatchV1.model_validate(payload)
+
+    # Ensure we can still deserialize strict v1 payloads even if they came as JSON.
+    raw = json.dumps({"version": 1, "ts": 1700000000000, "proposals": [], "metadata": {}}, ensure_ascii=False)
+    parsed = ProposalsBatchV1.model_validate_json(raw)
+    assert parsed.version == 1
+
+
+def test_strict_extra_fields_rejected() -> None:
+    with pytest.raises(ValidationError):
+        IntentV1.model_validate(
+            {
+                "version": 1,
+                "text": "x",
+                "lang_hint": "auto",
+                "screen": "System/Система",
+                "selection": {"kind": "none"},
+                "ts": 1700000000000,
+                "environment_mode": "FACTORY",
+                "snapshot_min": {},
+                "extra": "nope",
+            }
+        )