Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ State becomes:
```json
{
"facts": {
"focus.device": null
"focus.primary": null
},
"policies": {
"prohibit": ["docker"]
Expand Down Expand Up @@ -131,7 +131,7 @@ The compiler maintains an authoritative state snapshot:
```json
{
"facts": {
"focus.device": null
"focus.primary": null
},
"policies": {
"prohibit": []
Expand All @@ -141,7 +141,7 @@ The compiler maintains an authoritative state snapshot:
```

**Note**
In M1, the fact schema contains a single exclusive slot: `facts["focus.device"]`.
In M1, the fact schema contains a single exclusive slot: `facts["focus.primary"]`.

This slot exists to demonstrate deterministic fact replacement and correction semantics.
Richer fact schemas may be introduced in future milestones.
Expand Down Expand Up @@ -184,7 +184,7 @@ User: I'm using MacBook M3
State update:

```text
facts.focus.device = "MacBook M3"
facts.focus.primary = "MacBook M3"
```

Correction:
Expand All @@ -196,7 +196,7 @@ User: actually MacBook M2
Result:

```text
facts.focus.device = "MacBook M2"
facts.focus.primary = "MacBook M2"
```

Ambiguous mutation:
Expand Down
2 changes: 1 addition & 1 deletion docs/DescriptionAndMilestones.md
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ All state transitions originate from explicit user directives.
Explicit user commitments persist reliably within a conversation.
A correction means replacing a previously set fact, not evaluating conversational accuracy.

In M1 the fact schema intentionally contains a single exclusive slot (`facts["focus.device"]`).
In M1 the fact schema intentionally contains a single exclusive slot (`facts["focus.primary"]`).
This slot demonstrates deterministic fact replacement and correction semantics.

Policies (`policies.prohibit`) provide the primary mechanism for persistent conversational constraints.
Expand Down
10 changes: 5 additions & 5 deletions docs/M1Design.md
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ State is a deterministic snapshot:
`json
{
"facts": {
"focus.device": null
"focus.primary": null
},
"policies": {
"prohibit": []
Expand Down Expand Up @@ -120,7 +120,7 @@ Accepted patterns:

Produces:

FACT_SET(key="focus.device", value=X)
FACT_SET(key="focus.primary", value=X)

“Using” statements set the current discussion focus, not inventory.
Hard-positive directives store fact values as opaque strings and do not
Expand Down Expand Up @@ -228,8 +228,8 @@ While pending exists, no other mutation may occur.

#### Facts (exclusive)

focus.device = "nord stage 3"
focus.device = "nord stage 4"
focus.primary = "nord stage 3"
focus.primary = "nord stage 4"

→` only stage 4 active`

Expand All @@ -249,7 +249,7 @@ Example:

`json
{
"facts": {"focus.device": "nord stage 4"},
"facts": {"focus.primary": "nord stage 4"},
"policies": {"prohibit": ["parallel octaves"]}
}
`
Expand Down
3 changes: 2 additions & 1 deletion examples/01_persistent_guardrails.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,11 @@
from _util import print_json

from context_compiler import State, create_engine
from context_compiler.const import POLICY_PROHIBIT, STATE_POLICIES


def build_prompt(state: State, user_input: str) -> str:
prohibit = state["policies"]["prohibit"]
prohibit = state[STATE_POLICIES][POLICY_PROHIBIT]
prohibit_text = ", ".join(prohibit) if prohibit else "(none)"
return (
"System: Follow authoritative conversation state.\n"
Expand Down
4 changes: 3 additions & 1 deletion examples/04_tool_governance_denylist.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from _util import print_json

from context_compiler import create_engine
from context_compiler.const import POLICY_PROHIBIT, STATE_POLICIES


@dataclass
Expand Down Expand Up @@ -34,9 +35,10 @@ def main() -> None:
print()

print("Host-side tool denylist behavior:")
prohibit = state[STATE_POLICIES][POLICY_PROHIBIT]
tools = [Tool("docker"), Tool("kubectl")]
for tool in tools:
if tool.name in state["policies"]["prohibit"]:
if tool.name in prohibit:
block_tool(tool)
else:
allow_tool(tool)
Expand Down
14 changes: 14 additions & 0 deletions src/context_compiler/const.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
"""Schema key constants used by the context compiler state model."""

from typing import Final

# State schema
STATE_FACTS: Final = "facts"
STATE_POLICIES: Final = "policies"
STATE_VERSION: Final = "version"

# Fact keys
FOCUS_PRIMARY: Final = "focus.primary"

# Policy keys
POLICY_PROHIBIT: Final = "prohibit"
64 changes: 40 additions & 24 deletions src/context_compiler/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,19 @@
import re
from copy import deepcopy
from dataclasses import dataclass
from enum import StrEnum
from typing import Literal, TypedDict
from unicodedata import normalize as unicode_normalize

FactsState = TypedDict("FactsState", {"focus.device": str | None})
from .const import (
FOCUS_PRIMARY,
POLICY_PROHIBIT,
STATE_FACTS,
STATE_POLICIES,
STATE_VERSION,
)

FactsState = TypedDict("FactsState", {"focus.primary": str | None})


class PoliciesState(TypedDict):
Expand All @@ -26,8 +35,14 @@ class State(TypedDict):
version: Literal[1]


class DecisionKind(StrEnum):
UPDATE = "update"
PASSTHROUGH = "passthrough"
CLARIFY = "clarify"


class Decision(TypedDict):
kind: Literal["passthrough", "update", "clarify"]
kind: DecisionKind
state: State | None
prompt_to_user: str | None

Expand Down Expand Up @@ -61,7 +76,7 @@ class NegativeDirectiveRule:
_SPLIT_RE = re.compile(r",|\s+and\s+", re.IGNORECASE)

_PASSTHROUGH: Decision = {
"kind": "passthrough",
"kind": DecisionKind.PASSTHROUGH,
"state": None,
"prompt_to_user": None,
}
Expand Down Expand Up @@ -123,14 +138,15 @@ def _classify(self, user_input: str) -> PendingEvent | Decision:
if correction is not None:
if self._last_exclusive_fact_key is None:
return _clarify(
"I don't have a previous focus setting to correct. What should focus.device be?"
"I don't have a previous focus setting to correct. "
"What should focus.primary be?"
)
if not correction.strip():
return _clarify("What should focus.device be?")
return _clarify("What should focus.primary be?")
if _has_multiple_values(correction):
return _clarify("Corrections for focus.device must contain one value.")
return _clarify("Corrections for focus.primary must contain one value.")
if _invalid_exclusive_value(correction):
return _clarify("Corrections for focus.device must contain one value.")
return _clarify("Corrections for focus.primary must contain one value.")
return PendingEvent(kind="fact_set", fact_value=correction)

policy_add = _parse_hard_negative(normalized_message)
Expand All @@ -148,11 +164,11 @@ def _classify(self, user_input: str) -> PendingEvent | Decision:
positive_value = _parse_hard_positive(user_input, normalized_message)
if positive_value is not None:
if not positive_value.strip():
return _clarify("What should focus.device be?")
return _clarify("What should focus.primary be?")
if _has_multiple_values(positive_value):
return _clarify("focus.device accepts one value. Please provide a single value.")
return _clarify("focus.primary accepts one value. Please provide a single value.")
if _invalid_exclusive_value(positive_value):
return _clarify("focus.device value is unclear. Please provide one value.")
return _clarify("focus.primary value is unclear. Please provide one value.")
return PendingEvent(kind="fact_set", fact_value=positive_value)

pending = _parse_ambiguous_mutation(user_input)
Expand Down Expand Up @@ -197,7 +213,7 @@ def _apply_event(self, event: PendingEvent) -> Decision:
return self._remove_policies(list(event.values))
if event.kind == "fact_set":
assert event.fact_value is not None
return self._set_focus_device(event.fact_value)
return self._set_focus_primary(event.fact_value)
if event.kind == "reset_policies":
self._state = _initial_state()
self._last_exclusive_fact_key = None
Expand All @@ -207,45 +223,45 @@ def _apply_event(self, event: PendingEvent) -> Decision:
self._last_exclusive_fact_key = None
return _update_decision(self._state)

def _set_focus_device(self, value: str) -> Decision:
self._state["facts"]["focus.device"] = _clean_fact_value(value)
self._last_exclusive_fact_key = "focus.device"
def _set_focus_primary(self, value: str) -> Decision:
self._state[STATE_FACTS][FOCUS_PRIMARY] = _clean_fact_value(value)
self._last_exclusive_fact_key = FOCUS_PRIMARY
return _update_decision(self._state)

def _add_policies(self, values: list[str]) -> Decision:
existing = set(self._state["policies"]["prohibit"])
existing = set(self._state[STATE_POLICIES][POLICY_PROHIBIT])
for value in values:
existing.add(value)
self._state["policies"]["prohibit"] = sorted(existing)
self._state[STATE_POLICIES][POLICY_PROHIBIT] = sorted(existing)
return _update_decision(self._state)

def _remove_policies(self, values: list[str]) -> Decision:
existing = set(self._state["policies"]["prohibit"])
existing = set(self._state[STATE_POLICIES][POLICY_PROHIBIT])
for value in values:
existing.discard(value)
self._state["policies"]["prohibit"] = sorted(existing)
self._state[STATE_POLICIES][POLICY_PROHIBIT] = sorted(existing)
return _update_decision(self._state)


def _initial_state() -> State:
return {
"facts": {"focus.device": None},
"policies": {"prohibit": []},
"version": 1,
STATE_FACTS: {FOCUS_PRIMARY: None},
STATE_POLICIES: {POLICY_PROHIBIT: []},
STATE_VERSION: 1,
}


def _clarify(prompt: str) -> Decision:
return {
"kind": "clarify",
"kind": DecisionKind.CLARIFY,
"state": None,
"prompt_to_user": prompt,
}


def _update_decision(state: State) -> Decision:
return {
"kind": "update",
"kind": DecisionKind.UPDATE,
"state": deepcopy(state),
"prompt_to_user": None,
}
Expand Down Expand Up @@ -368,5 +384,5 @@ def _pending_prompt(pending: PendingEvent) -> str:
if pending.kind == "policy_remove" and pending.values:
return f"Did you mean to allow '{pending.values[0]}'?"
if pending.kind == "fact_set" and pending.fact_value is not None:
return f"Did you mean to set focus.device to '{pending.fact_value}'?"
return f"Did you mean to set focus.primary to '{pending.fact_value}'?"
return "Your directive was unclear. Did you mean to change state?"
Loading
Loading