diff --git a/README.md b/README.md index 418b5ef25f..91a616a603 100644 --- a/README.md +++ b/README.md @@ -2906,6 +2906,12 @@ snapshots and each doubling. These snapshots remain admitted beyond `queue_max_d and stay in memory until processing. Check extraction latency when warnings appear; process exit can still lose pending work. +Optional memory pre-screening and signal classification support TypeSafe (Jev) +and OpenAI Decisions. With `combine: auto`, their different endpoint paths keep +requests separate even behind the same gateway URL, model, and credentials. +Compatible requests to the same endpoint can still share one call. See the +[decision provider guide](backend/docs/GUARDRAILS.md). + The opt-in [DeerMem scope-isolation benchmark](backend/scripts/benchmark/deermem_scope_isolation/README.md) checks semantic safety across facts and summaries, and fact routing across users and agents. Failed extraction attempts are retryable execution errors, not safety passes. diff --git a/backend/docs/GUARDRAILS.md b/backend/docs/GUARDRAILS.md index 48a1be8d75..143a4402ef 100644 --- a/backend/docs/GUARDRAILS.md +++ b/backend/docs/GUARDRAILS.md @@ -544,6 +544,23 @@ make dev # then ask the agent to delete the disposable file with bash Check the denial message the agent receives and the `middleware:guardrail` journal event. Do not run this against real project data or a production key. +**The same gate over another endpoint: OpenAI Decisions.** `deerflow.guardrails.openai_decisions:OpenAIDecisionsGuardrailProvider` asks one `predicate` question about the call at `POST {base_url}/v1/decisions` and denies at the same `threshold`. It is the same *gate*, not a re-implementation: the strict-JSON local preflight, `max_state_chars` as a character count, the cache, the fail-closed rule and the whole transport (auth, retry, deadline, bounded body) are inherited from the Option 4 provider and `deerflow.typesafe`, so a change to the security boundary lands once. What differs is the envelope: + +| | TypeSafe (Jev) | OpenAI Decisions | +|---|---|---| +| Endpoint | `POST /v1/systemone` | `POST /v1/decisions` | +| Judged value | `state.tool_call` (name + argument JSON) | `input`: the same canonical argument JSON as text | +| Question | `noul` with a `criteria` map | `predicate`; the endpoint has no criteria field, so the rubric rides inside `instructions` | +| Refusal | not a response shape | `{"type": "refusal"}` — a per-question outcome, so the gate still denies for want of a verdict | +| `model` | defaults to `jev-latest` | **required**: no default model is shipped, so a deployment cannot send tool text to a model nobody named | + +`use:` is the only wiring — `guardrails.provider.config` keeps accepting the same fields, and there is no new config schema. The memory pre-screen and signal classification have siblings in the same shape (`deerflow.agents.memory.prescreen.openai_decisions:OpenAIDecisionsMemoryPrescreen`, `deerflow.agents.memory.signals.openai_decisions:OpenAIDecisionsSignalClassifier`); their thresholds, modes, cache buckets and `request_failed` reporting are the coordinator's and unchanged, and a `sharing_key` difference keeps a Decisions side from ever being combined into a Jev request. Offline suite: `tests/test_openai_decisions.py`. + +The endpoint path participates in the sharing key and public policy identity, so +`combine: auto` keeps Jev and Decisions requests separate even when their gateway +base URL, model, credentials, and transport match. Matching endpoints can still +combine. Missing credential and model errors name the configured service. + ## Implementing a Provider ### Required Interface diff --git a/backend/packages/harness/deerflow/agents/memory/prescreen/openai_decisions.py b/backend/packages/harness/deerflow/agents/memory/prescreen/openai_decisions.py new file mode 100644 index 0000000000..71f48b9306 --- /dev/null +++ b/backend/packages/harness/deerflow/agents/memory/prescreen/openai_decisions.py @@ -0,0 +1,58 @@ +"""OpenAI Decisions memory pre-screen: the same cost gate over ``/v1/decisions``. + +The gate's meaning is unchanged and lives in the base class: it answers "is this batch worth +paying an extraction call for?", it gates no execution and no write, and every failure direction +still extracts. Only the endpoint differs, so this subclass overrides the three things that +cannot be shared — which service the connection resolves against, which client carries it, and +what the provider is handed to judge — and inherits the rubric, the threshold, the cache and the +audit policy. + +Because the answers keep the shared ``NoulAnswer`` / ``AnswerSet`` shape, a deployment can move a +side from Jev to Decisions by editing its ``use:`` path alone, and the coordinator's combining, +cache-bucket and ``request_failed`` behaviour is untouched. + +Configuration:: + + memory: + backend_config: + prescreen: + mode: shadow + use: deerflow.agents.memory.prescreen.openai_decisions:OpenAIDecisionsMemoryPrescreen + config: + api_key_env: OPENAI_API_KEY + model: +""" + +from __future__ import annotations + +from deerflow.agents.memory.prescreen.contract import CONFIGURATION_SOURCE +from deerflow.agents.memory.prescreen.typesafe import TypeSafeMemoryPrescreen +from deerflow.openai_decisions.client import OpenAIDecisionsClient, resolve_decisions_connection +from deerflow.typesafe.client import TransportFactory, TypeSafeClient +from deerflow.typesafe.connection import TypeSafeConnection + + +class OpenAIDecisionsMemoryPrescreen(TypeSafeMemoryPrescreen): + """Skip the extraction call on a Decisions ``predicate`` probability below ``skip_threshold``.""" + + name = "openai_decisions" + policy_id = "deerflow.memory.prescreen.openai_decisions" + policy_version = "1.0.0" + + def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection: + return resolve_decisions_connection(settings=settings, configuration_source=CONFIGURATION_SOURCE) + + def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient: + return OpenAIDecisionsClient(connection, transport_factory=transport_factory) + + def _judge_state(self, batch_text: str) -> str: + """The batch text itself: this endpoint's ``input`` is text, not a state object. + + What is judged stays what the Jev side judges — the same + ``format_conversation_for_update`` output, bounded by the same ``max_state_chars`` + character count — only the envelope around it differs. + """ + return batch_text + + +__all__ = ["OpenAIDecisionsMemoryPrescreen"] diff --git a/backend/packages/harness/deerflow/agents/memory/prescreen/typesafe.py b/backend/packages/harness/deerflow/agents/memory/prescreen/typesafe.py index 6b4c59ad58..aa2bb14801 100644 --- a/backend/packages/harness/deerflow/agents/memory/prescreen/typesafe.py +++ b/backend/packages/harness/deerflow/agents/memory/prescreen/typesafe.py @@ -83,8 +83,8 @@ class TypeSafeMemoryPrescreen: if mode not in MODES: raise ValueError(f"memory.prescreen.mode must be one of {list(MODES)}, got {mode!r}") self._mode = mode - connection: TypeSafeConnection = resolve_connection( - settings={ + self._connection = self._resolve_connection( + { "api_key": api_key, "api_key_env": api_key_env, "base_url": base_url, @@ -93,12 +93,9 @@ class TypeSafeMemoryPrescreen: "deadline_seconds": deadline_seconds, "max_attempts": max_attempts, "retry_backoff": retry_backoff, - }, - defaults=typesafe_defaults(), - configuration_source=CONFIGURATION_SOURCE, + } ) - self._connection = connection - self._client = TypeSafeClient(connection, transport_factory=transport_factory) + self._client = self._build_client(self._connection, transport_factory=transport_factory) self.skip_threshold = finite_float("skip_threshold", skip_threshold, minimum=0.0, maximum=1.0) self._instructions = defaulted_text("instructions", instructions, _DEFAULT_INSTRUCTIONS) self._criteria_true = defaulted_text("criteria.true", criteria_entry(criteria, True), _DEFAULT_CRITERIA_TRUE) @@ -115,6 +112,19 @@ class TypeSafeMemoryPrescreen: } self._cache = AnswerCache(size=self.cache_size, ttl_seconds=self.cache_ttl_seconds) + # --- dialect hooks ------------------------------------------------------ + + def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection: + """Resolve this provider's effective connection: consumer config, then its own defaults.""" + return resolve_connection(settings=settings, defaults=typesafe_defaults(), configuration_source=CONFIGURATION_SOURCE) + + def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient: + return TypeSafeClient(connection, transport_factory=transport_factory) + + def _judge_state(self, batch_text: str) -> object: + """What the provider judges. A dialect whose endpoint takes plain input text overrides it.""" + return conversation_tail_state(batch_text) + # --- side interface (shared with the coordinator) --------------------- def questions(self) -> Mapping[str, Question]: @@ -123,7 +133,7 @@ class TypeSafeMemoryPrescreen: def ask(self, batch_text: str, questions: Mapping[str, Question]): """Send ``questions`` about ``batch_text`` through this side's client.""" - return self._client.ask(conversation_tail_state(batch_text), questions) + return self._client.ask(self._judge_state(batch_text), questions) def sharing_key(self, **dimensions: object) -> str: """Internal sharing identity (credential fingerprint, connection, limits, cache).""" @@ -175,7 +185,7 @@ class TypeSafeMemoryPrescreen: return self.interpret(cached.answers, model=cached.model_for(self._questions), cached=True) # No try/except: the transport failure is the coordinator's to record, and it # writes no bucket here either way (a failure must not be remembered). - answer_set = self._client.ask(conversation_tail_state(request.batch_text), self._questions) + answer_set = self._client.ask(self._judge_state(request.batch_text), self._questions) decision = self.interpret(answer_set.answers, model=answer_set.model, cached=False) if decision is not None: self._cache.put(request.digest, CachedVerdict(answers=dict(answer_set.answers), models={question_id: answer_set.model for question_id in answer_set.answers})) diff --git a/backend/packages/harness/deerflow/agents/memory/signals/openai_decisions.py b/backend/packages/harness/deerflow/agents/memory/signals/openai_decisions.py new file mode 100644 index 0000000000..95b441c87f --- /dev/null +++ b/backend/packages/harness/deerflow/agents/memory/signals/openai_decisions.py @@ -0,0 +1,47 @@ +"""OpenAI Decisions signal classifier: the same two additive hints over ``/v1/decisions``. + +Like its pre-screen sibling, this subclass changes the endpoint and nothing else. The two +directions, their rubrics, ``hint_threshold``, the label mapping, the per-question cache buckets +and the rule that a model hint can never write a confirmation all stay in the base class — the +reinforcement evidence gate keeps reading the deterministic set only. + +Configuration:: + + memory: + backend_config: + signal_classification: + mode: shadow + use: deerflow.agents.memory.signals.openai_decisions:OpenAIDecisionsSignalClassifier + config: + api_key_env: OPENAI_API_KEY + model: +""" + +from __future__ import annotations + +from deerflow.agents.memory.signals.contract import CONFIGURATION_SOURCE +from deerflow.agents.memory.signals.typesafe import TypeSafeSignalClassifier +from deerflow.openai_decisions.client import OpenAIDecisionsClient, resolve_decisions_connection +from deerflow.typesafe.client import TransportFactory, TypeSafeClient +from deerflow.typesafe.connection import TypeSafeConnection + + +class OpenAIDecisionsSignalClassifier(TypeSafeSignalClassifier): + """Map Decisions ``predicate`` probabilities for affirmation / negation onto hint labels.""" + + name = "openai_decisions" + policy_id = "deerflow.memory.signals.openai_decisions" + policy_version = "1.0.0" + + def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection: + return resolve_decisions_connection(settings=settings, configuration_source=CONFIGURATION_SOURCE) + + def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient: + return OpenAIDecisionsClient(connection, transport_factory=transport_factory) + + def _judge_state(self, batch_text: str) -> str: + """The batch text itself: this endpoint's ``input`` is text, not a state object.""" + return batch_text + + +__all__ = ["OpenAIDecisionsSignalClassifier"] diff --git a/backend/packages/harness/deerflow/agents/memory/signals/typesafe.py b/backend/packages/harness/deerflow/agents/memory/signals/typesafe.py index 33b6e9e5fa..aa3f729fd2 100644 --- a/backend/packages/harness/deerflow/agents/memory/signals/typesafe.py +++ b/backend/packages/harness/deerflow/agents/memory/signals/typesafe.py @@ -79,8 +79,8 @@ class TypeSafeSignalClassifier: if mode not in MODES: raise ValueError(f"memory.signal_classification.mode must be one of {list(MODES)}, got {mode!r}") self._mode = mode - connection: TypeSafeConnection = resolve_connection( - settings={ + self._connection = self._resolve_connection( + { "api_key": api_key, "api_key_env": api_key_env, "base_url": base_url, @@ -89,12 +89,9 @@ class TypeSafeSignalClassifier: "deadline_seconds": deadline_seconds, "max_attempts": max_attempts, "retry_backoff": retry_backoff, - }, - defaults=typesafe_defaults(), - configuration_source=CONFIGURATION_SOURCE, + } ) - self._connection = connection - self._client = TypeSafeClient(connection, transport_factory=transport_factory) + self._client = self._build_client(self._connection, transport_factory=transport_factory) self.hint_threshold = finite_float("hint_threshold", hint_threshold, minimum=0.0, maximum=1.0) affirmation_criteria = _side_criteria(criteria, QUESTION_AFFIRMATION) negation_criteria = _side_criteria(criteria, QUESTION_NEGATION) @@ -117,6 +114,19 @@ class TypeSafeSignalClassifier: } self._cache = AnswerCache(size=self.cache_size, ttl_seconds=self.cache_ttl_seconds) + # --- dialect hooks ------------------------------------------------------ + + def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection: + """Resolve this provider's effective connection: consumer config, then its own defaults.""" + return resolve_connection(settings=settings, defaults=typesafe_defaults(), configuration_source=CONFIGURATION_SOURCE) + + def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient: + return TypeSafeClient(connection, transport_factory=transport_factory) + + def _judge_state(self, batch_text: str) -> object: + """What the provider judges. A dialect whose endpoint takes plain input text overrides it.""" + return conversation_tail_state(batch_text) + # --- side interface (shared with the coordinator) --------------------- def questions(self) -> Mapping[str, Question]: @@ -125,7 +135,7 @@ class TypeSafeSignalClassifier: def ask(self, batch_text: str, questions: Mapping[str, Question]): """Send ``questions`` about ``batch_text`` through this side's client.""" - return self._client.ask(conversation_tail_state(batch_text), questions) + return self._client.ask(self._judge_state(batch_text), questions) def sharing_key(self, **dimensions: object) -> str: """Internal sharing identity (credential fingerprint, connection, limits, cache).""" @@ -192,7 +202,7 @@ class TypeSafeSignalClassifier: if not missing: assert cached is not None return self.interpret(answers, model=cached.model_for(self._questions), cached=True) - answer_set = self._client.ask(conversation_tail_state(request.batch_text), missing) + answer_set = self._client.ask(self._judge_state(request.batch_text), missing) answers.update(answer_set.answers) models.update({question_id: answer_set.model for question_id in answer_set.answers}) if answer_set.answers: diff --git a/backend/packages/harness/deerflow/guardrails/openai_decisions.py b/backend/packages/harness/deerflow/guardrails/openai_decisions.py new file mode 100644 index 0000000000..5124917227 --- /dev/null +++ b/backend/packages/harness/deerflow/guardrails/openai_decisions.py @@ -0,0 +1,71 @@ +"""OpenAI Decisions guardrail provider: the same pre-execution risk gate, another endpoint. + +Issue #7005 asked for a second decision backend behind the interfaces the Jev integration +already uses. Everything that makes this a *gate* is inherited from +:class:`~deerflow.guardrails.typesafe.TypeSafeGuardrailProvider` and deliberately not restated: +the strict-JSON local preflight, the character-count boundary, the threshold direction, the +cache, and fail-closed. Two copies of a security boundary drift apart, and the copy the primary +suite does not exercise is the one that goes quiet. + +What differs is the wire, and only the wire: + +* ``input`` is text, so the judged bytes are the canonical JSON the state digest already + identifies — the same bytes, not a second serialisation of them. +* a ``noul`` question becomes a ``predicate`` question, whose rubric rides in ``instructions`` + (see ``deerflow.openai_decisions.client``). +* a refusal is an answer, not an HTTP failure; the inherited ``_recorded_answer`` already treats + "no validated answer for my one question" as ``TypeSafeGuardrailError`` with cause + ``invalid_response``, so the gate denies on exactly the path it denies on today. + +Configuration is the existing slot — no new schema:: + + guardrails: + provider: + use: deerflow.guardrails.openai_decisions:OpenAIDecisionsGuardrailProvider + config: + api_key_env: OPENAI_API_KEY + model: +""" + +from __future__ import annotations + +from deerflow.guardrails.typesafe import TypeSafeGuardrailError, TypeSafeGuardrailProvider, _Probe +from deerflow.openai_decisions.client import OpenAIDecisionsClient, resolve_decisions_connection +from deerflow.typesafe.client import TransportFactory, TypeSafeClient +from deerflow.typesafe.connection import TypeSafeConnection + + +class OpenAIDecisionsGuardrailError(TypeSafeGuardrailError): + """A Decisions evaluation could not produce a usable verdict. + + Same family, same handling by ``guardrails.fail_closed``, different name so a log line and a + denial reason never attribute a decision to the wrong service. + """ + + +class OpenAIDecisionsGuardrailProvider(TypeSafeGuardrailProvider): + """Deny tool calls whose Decisions ``predicate`` probability reaches ``threshold``. + + Reads the same ``guardrails.provider.config`` block and the same optional fields as the Jev + provider; only ``base_url``/``model``/``api_key_env`` default differently, because they name a + different service. + """ + + name = "openai_decisions" + policy_id = "deerflow.guardrails.openai_decisions" + policy_version = "1.0.0" + code_prefix = "openai_decisions" + provider_name = "OpenAI Decisions" + error_class = OpenAIDecisionsGuardrailError + + def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection: + return resolve_decisions_connection(settings=settings, configuration_source=self.configuration_source) + + def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient: + return OpenAIDecisionsClient(connection, transport_factory=transport_factory) + + def _state(self, probe: _Probe) -> str: + return probe.state_text + + +__all__ = ["OpenAIDecisionsGuardrailError", "OpenAIDecisionsGuardrailProvider"] diff --git a/backend/packages/harness/deerflow/guardrails/typesafe.py b/backend/packages/harness/deerflow/guardrails/typesafe.py index 2271ca3541..251d634f49 100644 --- a/backend/packages/harness/deerflow/guardrails/typesafe.py +++ b/backend/packages/harness/deerflow/guardrails/typesafe.py @@ -102,6 +102,12 @@ class _Probe: tool_name: str arguments_text: str state_digest: str + state_text: str + """The exact canonical JSON the digest was taken over. + + Carried on the probe rather than re-derived: an endpoint dialect whose input is text instead + of a ``state`` object must judge the same bytes the digest identifies, and a second + serialisation of the same call could differ from the first.""" @dataclass(frozen=True) @@ -130,6 +136,21 @@ class TypeSafeGuardrailProvider: policy_id = "deerflow.guardrails.typesafe" policy_version = "1.1.0" + #: Prefix of every reason code this provider publishes. A dialect for another endpoint + #: overrides it so a denial names the service that actually decided. + code_prefix = "typesafe" + + #: Names this service in a message about a response that carried no answer at all. + provider_name = "TypeSafe" + + #: The config block whose values reached this provider, for the missing-credential message. + configuration_source = _CONFIGURATION_SOURCE + + #: Raised when an evaluation produced no usable verdict. A dialect overrides the class so a + #: failure names its own service, while staying inside the same exception family the + #: middleware's ``guardrails.fail_closed`` mapping already handles. + error_class = TypeSafeGuardrailError + def __init__( self, *, @@ -151,8 +172,8 @@ class TypeSafeGuardrailProvider: cache_ttl_seconds: float = 300.0, transport_factory: TransportFactory | None = None, ) -> None: - connection: TypeSafeConnection = resolve_connection( - settings={ + connection: TypeSafeConnection = self._resolve_connection( + { "api_key": api_key, "api_key_env": api_key_env, "base_url": base_url, @@ -161,12 +182,10 @@ class TypeSafeGuardrailProvider: "deadline_seconds": deadline_seconds, "max_attempts": max_attempts, "retry_backoff": retry_backoff, - }, - defaults=typesafe_defaults(), - configuration_source=_CONFIGURATION_SOURCE, + } ) self._connection = connection - self._client = TypeSafeClient(connection, transport_factory=transport_factory) + self._client = self._build_client(connection, transport_factory=transport_factory) self._threshold = finite_float("threshold", threshold, minimum=0.0, maximum=1.0) self._instructions = defaulted_text("instructions", instructions, DEFAULT_INSTRUCTIONS) self._criteria_true = defaulted_text("criteria.true", criteria_entry(criteria, True), DEFAULT_CRITERIA_TRUE) @@ -184,6 +203,18 @@ class TypeSafeGuardrailProvider: self._questions: dict[str, Question] = {_QUESTION_ID: self._question} self._cache: OrderedDict[tuple[str, str], _CacheEntry] = OrderedDict() + # --- dialect hooks ------------------------------------------------------ + + def _code(self, suffix: str) -> str: + return f"{self.code_prefix}.{suffix}" + + def _resolve_connection(self, settings: dict[str, object]) -> TypeSafeConnection: + """Resolve this provider's effective connection: provider config, then its own defaults.""" + return resolve_connection(settings=settings, defaults=typesafe_defaults(), configuration_source=self.configuration_source) + + def _build_client(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None) -> TypeSafeClient: + return TypeSafeClient(connection, transport_factory=transport_factory) + def release_policy_parameters(self) -> dict[str, object]: """Declare behaviour-affecting parameters for assembly identity (never the key). @@ -231,14 +262,14 @@ class TypeSafeGuardrailProvider: try: answer_set = self._client.ask(self._state(probe), self._questions) except TypeSafeError as exc: - raise TypeSafeGuardrailError(str(exc), cause=exc.cause) from exc + raise self.error_class(str(exc), cause=exc.cause) from exc return self._recorded_answer(answer_set) async def _ask_async(self, probe: _Probe) -> _Answer: try: answer_set = await self._client.aask(self._state(probe), self._questions) except TypeSafeError as exc: - raise TypeSafeGuardrailError(str(exc), cause=exc.cause) from exc + raise self.error_class(str(exc), cause=exc.cause) from exc return self._recorded_answer(answer_set) def _recorded_answer(self, answer_set: AnswerSet) -> _Answer: @@ -252,8 +283,8 @@ class TypeSafeGuardrailProvider: answer = answer_set.noul(_QUESTION_ID) if answer is None: error = answer_set.errors_by_question.get(_QUESTION_ID) - message = error.message if error is not None else f"TypeSafe response has no answer for question {_QUESTION_ID!r}" - raise TypeSafeGuardrailError(message, cause=CAUSE_INVALID_RESPONSE) + message = error.message if error is not None else f"{self.provider_name} response has no answer for question {_QUESTION_ID!r}" + raise self.error_class(message, cause=CAUSE_INVALID_RESPONSE) return _Answer(probability=answer.probability, model=recordable_model(answer_set.model)) def _state(self, probe: _Probe) -> dict[str, object]: @@ -273,14 +304,14 @@ class TypeSafeGuardrailProvider: if self._allowed_tools is not None and tool_name not in self._allowed_tools: return GuardrailDecision( allow=False, - reasons=[GuardrailReason(code="typesafe.tool_not_allowed", message=f"typesafe.tool_not_allowed: tool={tool_name!r} not in configured allowed_tools policy={self._policy_ref()}")], + reasons=[GuardrailReason(code=self._code("tool_not_allowed"), message=f"{self._code('tool_not_allowed')}: tool={tool_name!r} not in configured allowed_tools policy={self._policy_ref()}")], policy_id=self.policy_id, metadata={"tool_not_allowed": True}, ) if self._tools is not None and tool_name not in self._tools: return GuardrailDecision( allow=True, - reasons=[GuardrailReason(code="typesafe.tool_not_probed", message=f"typesafe.tool_not_probed: tool={tool_name!r} not in configured tools policy={self._policy_ref()}")], + reasons=[GuardrailReason(code=self._code("tool_not_probed"), message=f"{self._code('tool_not_probed')}: tool={tool_name!r} not in configured tools policy={self._policy_ref()}")], policy_id=self.policy_id, metadata={"tool_not_probed": True}, ) @@ -305,14 +336,14 @@ class TypeSafeGuardrailProvider: # when a consumer asks it to; it changes no limit). if len(arguments_text) > self._max_state_chars: return self._state_unusable("too_large", limit=self._max_state_chars, length=len(arguments_text), digest=state_digest) - return _Probe(tool_name=tool_name, arguments_text=arguments_text, state_digest=state_digest) + return _Probe(tool_name=tool_name, arguments_text=arguments_text, state_digest=state_digest, state_text=state_text) def _state_unusable(self, cause: str, **fields: object) -> GuardrailDecision: """Deny a call TypeSafe must not see: nothing truncated, nothing lossy is sent.""" detail = " ".join(f"{key}={value}" for key, value in fields.items()) return GuardrailDecision( allow=False, - reasons=[GuardrailReason(code="typesafe.state_unusable", message=f"typesafe.state_unusable: cause={cause} {detail} policy={self._policy_ref()}")], + reasons=[GuardrailReason(code=self._code("state_unusable"), message=f"{self._code('state_unusable')}: cause={cause} {detail} policy={self._policy_ref()}")], policy_id=self.policy_id, metadata={"state_unusable_cause": cause, **fields}, ) @@ -326,7 +357,7 @@ class TypeSafeGuardrailProvider: return self._model_decision(allow=entry.allow, probability=entry.probability, model=entry.model, state_digest=entry.state_digest, cached=True) def _model_decision(self, *, allow: bool, probability: float, model: str, state_digest: str, cached: bool) -> GuardrailDecision: - code = "typesafe.allowed" if allow else "typesafe.tool_call_risky" + code = self._code("allowed" if allow else "tool_call_risky") # repr(): the recorded probability and threshold must replay to the same # verdict, so neither is rounded for display. repr() round-trips exactly. # ``model`` is already bounded by recordable_model: it is response diff --git a/backend/packages/harness/deerflow/openai_decisions/__init__.py b/backend/packages/harness/deerflow/openai_decisions/__init__.py new file mode 100644 index 0000000000..52a6708a2b --- /dev/null +++ b/backend/packages/harness/deerflow/openai_decisions/__init__.py @@ -0,0 +1,45 @@ +"""OpenAI Decisions transport for the host-side decision consumers. + +Sibling of ``deerflow.typesafe`` for issue #7005: the same three consumer slots, the same +policy ownership, a different endpoint dialect. + +* ``guardrails/openai_decisions.py`` — the pre-execution tool risk gate. +* ``agents/memory/prescreen/openai_decisions.py`` — the memory capture pre-screen. +* ``agents/memory/signals/openai_decisions.py`` — signal classification. + +Shared here: the endpoint, the request envelope, question encoding, answer validation, and the +provider's own defaults. Transport lifecycle, authentication, retry, the deadline budget and +the bounded body read are inherited from ``deerflow.typesafe.client.TypeSafeClient`` rather than +restated, so one owner holds those rules for both backends. + +The split of what stays in each adapter is `deerflow.typesafe`'s, unchanged: state content, +questions, criteria, thresholds, failure policy, cache semantics and business audit. +""" + +from deerflow.openai_decisions.client import ( + ANSWER_TYPE_REFUSAL, + CATEGORY_REFUSAL, + DEFAULT_API_KEY_ENV, + DEFAULT_BASE_URL, + ENDPOINT_PATH, + QUESTION_CHOICE, + QUESTION_PREDICATE, + OpenAIDecisionsClient, + encode_question, + predicate_instructions, + provider_defaults, +) + +__all__ = [ + "ANSWER_TYPE_REFUSAL", + "CATEGORY_REFUSAL", + "DEFAULT_API_KEY_ENV", + "DEFAULT_BASE_URL", + "ENDPOINT_PATH", + "OpenAIDecisionsClient", + "QUESTION_CHOICE", + "QUESTION_PREDICATE", + "encode_question", + "predicate_instructions", + "provider_defaults", +] diff --git a/backend/packages/harness/deerflow/openai_decisions/client.py b/backend/packages/harness/deerflow/openai_decisions/client.py new file mode 100644 index 0000000000..ddf84766a3 --- /dev/null +++ b/backend/packages/harness/deerflow/openai_decisions/client.py @@ -0,0 +1,222 @@ +"""OpenAI Decisions transport: DeerFlow's question set spoken as one ``/v1/decisions`` request. + +Issue #7005 asked for a second decision backend behind the interfaces the TypeSafe (Jev) +integration already uses, not a new decision system. The provider slots +(``guardrails.provider.use``, ``memory.prescreen.use``, ``memory.signal_classification.use``) +already resolve a class path and hand it the consumer's ``config``, so a new backend is a +sibling adapter and the workflows, thresholds and failure directions stay where they are. + +Ownership follows `deerflow.typesafe`: this module owns the wire — endpoint, envelope, how a +question is encoded, how an answer is validated, and the error taxonomy — and nothing else. +Thresholds, rubrics, caches and the direction a failure takes stay in each adapter, so the +memory pre-screen still means "skip the extraction call" here exactly as it means there. +Transport itself (authentication, retry, backoff, the deadline budget, the bounded and +never-decoded body read) is inherited unchanged from +:class:`~deerflow.typesafe.client.TypeSafeClient`; re-declaring it in a second module would put +those security properties where they can drift. + +Two differences are structural, and the mapping below exists because of them: + +* **``input`` is the judged text, not a state object.** Jev takes a ``state`` mapping whose + shape each consumer owns. Here the consumer decides the same thing through its + ``_judge_state`` hook, so nothing in this module knows what a tool call or a conversation + batch is. +* **A ``predicate`` question has no field for the two criteria.** A ``noul`` question's + ``criteria`` say when the answer is true and when it is false; the Decisions endpoint only + carries ``instructions``. Folding them into the instruction text keeps one rubric per + question instead of silently dropping half of it, which would move the decision boundary + while the adapter went on reporting the same ``criteria`` in its policy identity. + +A refusal is an *answer* on this endpoint (``{"type": "refusal"}``), not an HTTP failure. +It therefore becomes a question-level :class:`~deerflow.typesafe.client.QuestionError` — the +same channel a malformed answer uses — so the gate denies for want of a verdict while the +memory paths carry on, which is each consumer's existing policy rather than a new branch. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping + +from deerflow.typesafe.client import ( + CATEGORY_MISSING, + CATEGORY_TYPE, + Answer, + AnswerSet, + Question, + QuestionError, + TypeSafeClient, + validate_choice_label, + validate_probability, +) +from deerflow.typesafe.connection import TypeSafeConnection, resolve_connection +from deerflow.typesafe.errors import CAUSE_INVALID_RESPONSE, TypeSafeError + +DEFAULT_BASE_URL = "https://api.openai.com" +DEFAULT_API_KEY_ENV = "OPENAI_API_KEY" +ENDPOINT_PATH = "/v1/decisions" + +#: The Decisions spelling of a bounded yes/no judgement, which is what ``noul`` asks for. +QUESTION_PREDICATE = "predicate" +QUESTION_CHOICE = "choice" + +#: ``refusal`` is a value of an answer's ``type``, so it is a per-question outcome. +ANSWER_TYPE_REFUSAL = "refusal" +CATEGORY_REFUSAL = "refusal" + + +def provider_defaults() -> dict[str, object]: + """The fallback layer this backend passes to ``resolve_connection``. + + ``model`` is deliberately blank rather than a model id. Which decision model answers is an + operator choice with a cost and a data-handling implication, and a checked-in default would + send conversation and tool text to a model nobody selected. The resolver rejects a blank + model at construction, so a deployment naming this backend has to name a model too. + """ + return {"base_url": DEFAULT_BASE_URL, "api_key_env": DEFAULT_API_KEY_ENV, "model": ""} + + +def resolve_decisions_connection(*, settings: Mapping[str, object] | None = None, defaults: Mapping[str, object] | None = None, configuration_source: str) -> TypeSafeConnection: + """Resolve a connection that speaks this endpoint, or raise at construction. + + One owner pairs the provider's defaults with its request path: a "Decisions" connection + resolved without ``ENDPOINT_PATH`` would post to the other service's URL, and nothing but a + live request would say so. ``defaults`` is layered *over* the provider's own defaults, which + are in turn overridden by ``settings`` — the precedence each adapter already documents. + """ + return resolve_connection( + settings=settings, + defaults={**provider_defaults(), **(defaults or {})}, + configuration_source=configuration_source, + endpoint_path=ENDPOINT_PATH, + provider_label="OpenAI Decisions", + ) + + +class OpenAIDecisionsClient(TypeSafeClient): + """One consumer's Decisions client, built from its effective connection. + + Inherits the whole transport lifecycle; overrides only the request skeleton and the + response parsing, which are the parts that are protocol-specific. + """ + + #: ``noul``/System One wording in an inherited message would attribute a decision to the + #: wrong service, and these messages reach logs, denial reasons and the evaluation report. + provider_label = "OpenAI Decisions" + + def _payload(self, state: object, questions: Mapping[str, Question]) -> dict[str, object]: + return { + "model": self._connection.model, + "input": state, + "questions": [encode_question(question_id, question) for question_id, question in questions.items()], + } + + def _parse(self, body: bytes, questions: Mapping[str, Question]) -> AnswerSet: + """Split one response into an envelope plus per-question outcomes, as ``typesafe`` does.""" + try: + payload = json.loads(body) + # ``RecursionError`` and not ``ValueError`` is what deep nesting raises, and it must stay + # inside this taxonomy: a hostile or broken endpoint must not escape as an unrelated + # exception, which the consumer would lose the round's record over. + except (ValueError, RecursionError) as exc: + raise TypeSafeError(f"{self.provider_label} response was not valid JSON", cause=CAUSE_INVALID_RESPONSE) from exc + if not isinstance(payload, dict): + raise TypeSafeError(f"{self.provider_label} response must be a JSON object", cause=CAUSE_INVALID_RESPONSE) + raw_answers = payload.get("answers") + if not isinstance(raw_answers, list): + raise TypeSafeError(f"{self.provider_label} response is missing an 'answers' array", cause=CAUSE_INVALID_RESPONSE) + + by_name = _answers_by_name(raw_answers) + # ``model`` is optional here: the endpoint may echo the served version, and when it does + # that value is what a consumer records. When it does not, the model this request named is + # recorded instead — provenance is still needed for the audit trail, and unlike a value + # the endpoint returned it cannot have been chosen to be echoed into a log. + model = payload.get("model") + served = model if isinstance(model, str) and model else self._connection.model + + validated: dict[str, Answer] = {} + errors: dict[str, QuestionError] = {} + for question_id, question in questions.items(): + answer, error = validate_answer(question_id, question, by_name.get(question_id), provider_label=self.provider_label) + if error is not None: + errors[question_id] = error + elif answer is not None: + validated[question_id] = answer + return AnswerSet(model=served, answers=validated, errors_by_question=errors) + + +def encode_question(question_id: str, question: Question) -> dict[str, object]: + """Encode one shared :class:`Question` as a Decisions question object.""" + if question.type == QUESTION_CHOICE: + # A choice question does carry its options, so the criteria map onto them directly and the + # instruction text is left alone; silently narrowing it would change what is judged. + return { + "type": QUESTION_CHOICE, + "name": question_id, + "instructions": question.instructions, + "choices": [{"value": label, "description": description} for label, description in (question.criteria or {}).items()], + } + return {"type": QUESTION_PREDICATE, "name": question_id, "instructions": predicate_instructions(question)} + + +def predicate_instructions(question: Question) -> str: + """The rubric a predicate question has no field for, folded into its instructions.""" + if not question.criteria: + return question.instructions + rubric = "".join(f"\nAnswer {label} when: {description}" for label, description in question.criteria.items()) + return f"{question.instructions}{rubric}" + + +def _answers_by_name(raw_answers: list[object]) -> dict[str, dict[object, object]]: + """Index the answer array by ``name``, first occurrence winning. + + An array can repeat a name while a mapping cannot, and which of two conflicting answers a + consumer would act on must not depend on how a server orders them. Answers without a usable + name, and names nobody asked about, are ignored rather than reported: the shared contract is + that only the questions actually asked appear in the outcome. + """ + by_name: dict[str, dict[object, object]] = {} + for entry in raw_answers: + if not isinstance(entry, dict): + continue + name = entry.get("name") + if isinstance(name, str) and name not in by_name: + by_name[name] = entry + return by_name + + +def validate_answer(question_id: str, question: Question, entry: dict[object, object] | None, *, provider_label: str) -> tuple[Answer | None, QuestionError | None]: + """Validate one answer against the question that was asked, reusing the shared rules.""" + if entry is None: + return None, QuestionError(question_id=question_id, category=CATEGORY_MISSING, message=f"{provider_label} response has no answer for question {question_id!r}") + answer_type = entry.get("type") + if answer_type == ANSWER_TYPE_REFUSAL: + # Only the outcome is reported. A refusal object is response content and can carry text the + # endpoint chose, which must not reach a log or a reason message unchecked. + return None, QuestionError(question_id=question_id, category=CATEGORY_REFUSAL, message=f"{provider_label} refused to answer question {question_id!r}") + expected = QUESTION_CHOICE if question.type == QUESTION_CHOICE else QUESTION_PREDICATE + if answer_type != expected: + return None, QuestionError( + question_id=question_id, + category=CATEGORY_TYPE, + message=f"{provider_label} answer for {question_id!r} is not a {expected} answer (type={type(answer_type).__name__})", + ) + if expected == QUESTION_PREDICATE: + return validate_probability(question_id, entry.get("probability"), provider_label=provider_label, field_name="predicate") + return validate_choice_label(question_id, question, entry.get("choice"), provider_label=provider_label) + + +__all__ = [ + "ANSWER_TYPE_REFUSAL", + "CATEGORY_REFUSAL", + "DEFAULT_API_KEY_ENV", + "DEFAULT_BASE_URL", + "ENDPOINT_PATH", + "QUESTION_CHOICE", + "QUESTION_PREDICATE", + "OpenAIDecisionsClient", + "encode_question", + "predicate_instructions", + "provider_defaults", + "resolve_decisions_connection", +] diff --git a/backend/packages/harness/deerflow/typesafe/AGENTS.md b/backend/packages/harness/deerflow/typesafe/AGENTS.md index 86bffe2726..a6d2238058 100644 --- a/backend/packages/harness/deerflow/typesafe/AGENTS.md +++ b/backend/packages/harness/deerflow/typesafe/AGENTS.md @@ -50,3 +50,22 @@ Credential resolution is per *layer* too — the first layer that sets `api_key` consumer naming an environment variable is never silently switched to another layer's key. Likewise `base_url` must be an `http(s)` URL with no query, fragment or embedded credentials, checked at construction rather than on every request. + +**Dialects.** `deerflow/openai_decisions/` (issue #7005) speaks a second endpoint over this same +client rather than copying it, because the rules above are security properties and a second copy +of them can drift unnoticed. Two seams carry a dialect, and they are the only two: +``TypeSafeClient.provider_label``, which names the service in every request-level message (a +failure must not tell an operator about a service they did not configure), and +``TypeSafeConnection.endpoint_path``, which the connection composes into ``url`` so an endpoint +path belongs to the backend that resolves the connection rather than to the base URL it happens +to share. The endpoint path is included in ``public_parameters()`` and therefore in both the +sharing key and public policy identity: different dialects behind one gateway cannot share a +combined memory request. Matching endpoints may still combine when the other settings match. +``resolve_connection(provider_label=...)`` also names the configured service in credential +and model construction errors. ``openai_decisions.client.resolve_decisions_connection`` pairs +the endpoint and service label, so a +"Decisions" connection cannot be built pointing at System One. A dialect adds no threshold, no +rubric and no failure policy: the three adapters override their connection resolution, their +client and the value the endpoint judges (``input`` text instead of a ``state`` object), and +inherit the rest. + diff --git a/backend/packages/harness/deerflow/typesafe/client.py b/backend/packages/harness/deerflow/typesafe/client.py index 46dca7ab5d..fc12a9238f 100644 --- a/backend/packages/harness/deerflow/typesafe/client.py +++ b/backend/packages/harness/deerflow/typesafe/client.py @@ -172,7 +172,7 @@ type Answer = NoulAnswer | ChoiceAnswer class QuestionError: """A question-level failure: the request was usable, this question's answer was not. - ``category`` is one of ``missing``, ``type``, ``probability`` or ``label``. + ``category`` is one of ``missing``, ``type``, ``probability``, ``label`` or ``refusal``. ``message`` names the problem without repeating the response body, which a malformed or hostile endpoint can fill with echoed state. """ @@ -217,6 +217,11 @@ class TypeSafeClient: §3.1). """ + #: Names the endpoint in every request-level message this client raises. A + #: dialect subclass that speaks another provider's protocol overrides it, so a + #: failure never reports the wrong operator's service. + provider_label = "TypeSafe" + def __init__(self, connection: TypeSafeConnection, *, transport_factory: TransportFactory | None = None) -> None: self._connection = connection self._transport_factory = transport_factory @@ -333,9 +338,9 @@ class TypeSafeClient: # request it failed to build — for a rejected header that is the whole # ``Bearer `` value — and the guardrail's ``logger.exception`` # prints the cause chain. The type name below is kept for diagnosis. - raise _RetryableAttempt(f"TypeSafe request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None + raise _RetryableAttempt(f"{self.provider_label} request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None except httpx.HTTPError as exc: - raise TypeSafeError(f"TypeSafe request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None + raise TypeSafeError(f"{self.provider_label} request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None async def _attempt_async(self, client: httpx.AsyncClient, payload: dict[str, object], headers: dict[str, str], questions: Mapping[str, Question], deadline_at: float) -> AnswerSet: try: @@ -352,9 +357,9 @@ class TypeSafeClient: except httpx.TransportError as exc: # ``from None`` for the same reason as the sync path: a rejected header # would otherwise carry the ``Bearer`` value in the printed chain. - raise _RetryableAttempt(f"TypeSafe request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None + raise _RetryableAttempt(f"{self.provider_label} request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None except httpx.HTTPError as exc: - raise TypeSafeError(f"TypeSafe request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None + raise TypeSafeError(f"{self.provider_label} request failed: {type(exc).__name__}", cause=CAUSE_TRANSPORT) from None def _status_error(self, status_code: int) -> TypeSafeError: # Only the numeric status is reported. The response body *and* the HTTP @@ -362,7 +367,7 @@ class TypeSafeClient: # (raw tool arguments, conversation text) in either, and this message # reaches middleware logs and the evaluation report. hint = " (check the API key configured via api_key / the API key environment variable)" if status_code == _UNAUTHORIZED_STATUS else "" - message = f"TypeSafe returned HTTP {status_code}{hint}" + message = f"{self.provider_label} returned HTTP {status_code}{hint}" if status_code in _RETRYABLE_STATUS_CODES: return _RetryableAttempt(message, cause=CAUSE_HTTP_STATUS) return TypeSafeError(message, cause=CAUSE_HTTP_STATUS) @@ -378,15 +383,15 @@ class TypeSafeClient: # would escape this module's taxonomy and the consumer would lose the round's # record instead of falling back. except (ValueError, RecursionError) as exc: - raise TypeSafeError("TypeSafe response was not valid JSON", cause=CAUSE_INVALID_RESPONSE) from exc + raise TypeSafeError(f"{self.provider_label} response was not valid JSON", cause=CAUSE_INVALID_RESPONSE) from exc if not isinstance(payload, dict): - raise TypeSafeError("TypeSafe response must be a JSON object", cause=CAUSE_INVALID_RESPONSE) + raise TypeSafeError(f"{self.provider_label} response must be a JSON object", cause=CAUSE_INVALID_RESPONSE) model = payload.get("model") if not isinstance(model, str) or not model: - raise TypeSafeError("TypeSafe response is missing a non-empty 'model'", cause=CAUSE_INVALID_RESPONSE) + raise TypeSafeError(f"{self.provider_label} response is missing a non-empty 'model'", cause=CAUSE_INVALID_RESPONSE) answers = payload.get("answers") if not isinstance(answers, dict): - raise TypeSafeError("TypeSafe response is missing an 'answers' object", cause=CAUSE_INVALID_RESPONSE) + raise TypeSafeError(f"{self.provider_label} response is missing an 'answers' object", cause=CAUSE_INVALID_RESPONSE) validated: dict[str, Answer] = {} errors: dict[str, QuestionError] = {} @@ -406,7 +411,7 @@ class TypeSafeClient: raise self._deadline_error() def _deadline_error(self) -> TypeSafeError: - return TypeSafeError(f"TypeSafe evaluation exceeded deadline_seconds={self._connection.deadline_seconds:g}", cause=CAUSE_DEADLINE) + return TypeSafeError(f"{self.provider_label} evaluation exceeded deadline_seconds={self._connection.deadline_seconds:g}", cause=CAUSE_DEADLINE) def _remaining(self, deadline_at: float) -> float: remaining = deadline_at - time.monotonic() @@ -491,32 +496,49 @@ def _validate_answer(question_id: str, question: Question, raw: object) -> tuple return _validate_choice(question_id, question, raw) -def _validate_noul(question_id: str, raw: dict[object, object]) -> tuple[Answer | None, QuestionError | None]: - raw_probability = raw.get("noul") - # bool is an int subclass, and json.loads parses NaN/Infinity literals, so both - # are rejected here rather than reaching a consumer's comparison. Only the - # offending type is reported, for the reason in _validate_answer. +def validate_probability(question_id: str, raw_probability: object, *, provider_label: str, field_name: str) -> tuple[Answer | None, QuestionError | None]: + """One owner for the probability rules every dialect's numeric answer needs. + + ``bool`` is an ``int`` subclass and ``json.loads`` accepts the ``NaN``/``Infinity`` + literals, so both are rejected here rather than reaching a consumer's threshold + comparison. Only the offending type is reported, for the reason in + :func:`_validate_answer`. A ``QuestionError`` is returned instead of raising so a + bad answer cannot discard the other valid answers in the same response (design §2.3). + """ if isinstance(raw_probability, bool) or not isinstance(raw_probability, (int, float)): - return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message=f"TypeSafe returned a non-numeric noul probability (type={type(raw_probability).__name__})") + return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message=f"{provider_label} returned a non-numeric {field_name} probability (type={type(raw_probability).__name__})") try: probability = float(raw_probability) except OverflowError: # A JSON integer too large for float() (e.g. 10**400) must stay a # per-question error: letting it escape would discard the other valid # answers in the same response (design §2.3). - return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message="TypeSafe returned a noul probability outside [0, 1]") + return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message=f"{provider_label} returned a {field_name} probability outside [0, 1]") if not math.isfinite(probability) or not 0.0 <= probability <= 1.0: - return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message="TypeSafe returned a noul probability outside [0, 1]") + return None, QuestionError(question_id=question_id, category=CATEGORY_PROBABILITY, message=f"{provider_label} returned a {field_name} probability outside [0, 1]") return NoulAnswer(probability=probability), None +def _validate_noul(question_id: str, raw: dict[object, object]) -> tuple[Answer | None, QuestionError | None]: + return validate_probability(question_id, raw.get("noul"), provider_label="TypeSafe", field_name="noul") + + +def validate_choice_label(question_id: str, question: Question, raw_label: object, *, provider_label: str) -> tuple[Answer | None, QuestionError | None]: + """One owner for the rule that a ``choice`` answer must be one of the labels asked for. + + The label set is the question's own criteria: an answer outside it is a + question-level failure, never a verdict, because a consumer maps labels to + behaviour and an unrecognised one has no mapping to apply. + """ + if not isinstance(raw_label, str) or not raw_label: + return None, QuestionError(question_id=question_id, category=CATEGORY_LABEL, message=f"{provider_label} returned a non-textual choice label (type={type(raw_label).__name__})") + if question.criteria is None or raw_label not in question.criteria: + return None, QuestionError(question_id=question_id, category=CATEGORY_LABEL, message=f"{provider_label} returned a choice label outside the configured criteria") + return ChoiceAnswer(label=raw_label), None + + def _validate_choice(question_id: str, question: Question, raw: dict[object, object]) -> tuple[Answer | None, QuestionError | None]: - label = raw.get("choice") - if not isinstance(label, str) or not label: - return None, QuestionError(question_id=question_id, category=CATEGORY_LABEL, message=f"TypeSafe returned a non-textual choice label (type={type(label).__name__})") - if question.criteria is None or label not in question.criteria: - return None, QuestionError(question_id=question_id, category=CATEGORY_LABEL, message="TypeSafe returned a choice label outside the configured criteria") - return ChoiceAnswer(label=label), None + return validate_choice_label(question_id, question, raw.get("choice"), provider_label="TypeSafe") def _factory_identity(factory: TransportFactory | None) -> str | None: @@ -552,5 +574,7 @@ __all__ = [ "TransportFactory", "TypeSafeClient", "recordable_model", + "validate_choice_label", + "validate_probability", "wire_size", ] diff --git a/backend/packages/harness/deerflow/typesafe/connection.py b/backend/packages/harness/deerflow/typesafe/connection.py index 1260603686..5fe6a55dbb 100644 --- a/backend/packages/harness/deerflow/typesafe/connection.py +++ b/backend/packages/harness/deerflow/typesafe/connection.py @@ -81,11 +81,15 @@ class TypeSafeConnection: deadline_seconds: float max_attempts: int retry_backoff: float + #: Which request path the endpoint serves. It belongs to the *dialect*, not the operator: + #: a connection whose base URL points at another service must not report a System One URL, so + #: every consumer resolves it through the backend it is about to call. + endpoint_path: str = ENDPOINT_PATH @property def url(self) -> str: - """The System One endpoint this connection posts to.""" - return f"{self.base_url.rstrip('/')}{ENDPOINT_PATH}" + """The endpoint this connection posts to.""" + return f"{self.base_url.rstrip('/')}{self.endpoint_path}" def credential_fingerprint(self) -> str: """A short digest of the credential, to compare consumers without recording either key.""" @@ -96,6 +100,7 @@ class TypeSafeConnection: return { "model": self.model, "base_url": self.base_url.rstrip("/"), + "endpoint_path": self.endpoint_path, "timeout": self.timeout, "deadline_seconds": self.deadline_seconds, "max_attempts": self.max_attempts, @@ -108,6 +113,8 @@ def resolve_connection( settings: Mapping[str, object] | None = None, defaults: Mapping[str, object] | None = None, configuration_source: str, + endpoint_path: str = ENDPOINT_PATH, + provider_label: str = "TypeSafe", ) -> TypeSafeConnection: """Resolve the effective connection for one consumer. @@ -122,7 +129,13 @@ def resolve_connection( ``configuration_source`` names the consumer's settings location (for example ``guardrails.provider.config``); it reaches the missing-credential message so an operator knows which block to edit. It never carries the credential. + + ``endpoint_path`` is what the calling backend serves, not something a consumer + configures; it is validated here so a malformed value cannot become a per-request URL error. + ``provider_label`` names that backend in construction-time diagnostics. """ + if not isinstance(endpoint_path, str) or not endpoint_path.startswith("/"): + raise ValueError("endpoint_path must be an absolute request path such as '/v1/systemone'") override = _configured(settings) fallback = _configured(defaults) @@ -132,7 +145,7 @@ def resolve_connection( api_key = _resolve_credential(override, fallback, api_key_env) if not isinstance(api_key, str) or not api_key: - raise ValueError(f"TypeSafe requires an API key: pass 'api_key' in {configuration_source} or set the {api_key_env} environment variable") + raise ValueError(f"{provider_label} requires an API key: pass 'api_key' in {configuration_source} or set the {api_key_env} environment variable") # Shape only, and never echoed: a key that cannot travel as a header value is a # configuration error here instead of a per-call protocol error whose message # carries the whole ``Bearer`` header into the guardrail's exception log. @@ -142,7 +155,7 @@ def resolve_connection( model = _first(override, fallback, "model", DEFAULT_MODEL) if not isinstance(model, str) or not model.strip(): - raise ValueError("model must be a non-empty string; TypeSafe returns the version it actually served") + raise ValueError(f"{provider_label} model must be a non-empty string") return TypeSafeConnection( api_key=api_key, @@ -153,6 +166,7 @@ def resolve_connection( deadline_seconds=finite_float("deadline_seconds", _first(override, fallback, "deadline_seconds", DEFAULT_DEADLINE_SECONDS), minimum=0.0, exclusive=True), max_attempts=whole_number("max_attempts", _first(override, fallback, "max_attempts", DEFAULT_MAX_ATTEMPTS), minimum=1), retry_backoff=finite_float("retry_backoff", _first(override, fallback, "retry_backoff", DEFAULT_RETRY_BACKOFF), minimum=0.0), + endpoint_path=endpoint_path, ) @@ -162,6 +176,7 @@ def resolve_connection_for_mode( settings: Mapping[str, object] | None = None, defaults: Mapping[str, object] | None = None, configuration_source: str, + endpoint_path: str = ENDPOINT_PATH, ) -> TypeSafeConnection | None: """Resolve a mode-bearing consumer's connection, or ``None`` when it is off. @@ -175,7 +190,7 @@ def resolve_connection_for_mode( """ if mode == MODE_OFF: return None - return resolve_connection(settings=settings, defaults=defaults, configuration_source=configuration_source) + return resolve_connection(settings=settings, defaults=defaults, configuration_source=configuration_source, endpoint_path=endpoint_path) def _configured(settings: Mapping[str, object] | None) -> dict[str, object]: diff --git a/backend/tests/test_openai_decisions.py b/backend/tests/test_openai_decisions.py new file mode 100644 index 0000000000..92e85496c3 --- /dev/null +++ b/backend/tests/test_openai_decisions.py @@ -0,0 +1,552 @@ +"""Tests for the OpenAI Decisions backend (issue #7005). + +Three things are pinned here, and each is a claim the change could silently break: + +* **The wire mapping both directions.** A ``noul`` question must leave as a ``predicate`` + question whose rubric still travels with it, and an answer array must come back as the same + ``NoulAnswer`` / ``ChoiceAnswer`` / ``QuestionError`` shapes the Jev client produces — because + every consumer's threshold, cache bucket and failure direction is written against those. +* **That the shared transport rules still hold for a second dialect.** Bounded never-decoded + body, the two-layer error split, the identity encoding, and "a refusal is a per-question + outcome". A duplicated transport would have had to re-earn each of those. +* **That the gate's direction did not move.** The tool gate still denies at or above its + threshold and fails closed on a refusal, while the memory sides still fall back — the same + request, opposite policies, owned by each adapter as before. + +Everything runs offline through ``httpx.MockTransport``, the pattern the Jev suites use. +""" + +from __future__ import annotations + +import gzip +import hashlib +import json + +import httpx +import pytest + +from deerflow.agents.memory.prescreen.contract import MODE_ENFORCE, resolve_memory_prescreen +from deerflow.agents.memory.prescreen.openai_decisions import OpenAIDecisionsMemoryPrescreen +from deerflow.agents.memory.prescreen.typesafe import QUESTION_ID as PRESCREEN_QUESTION +from deerflow.agents.memory.prescreen.typesafe import TypeSafeMemoryPrescreen +from deerflow.agents.memory.signals.contract import LABEL_REINFORCEMENT +from deerflow.agents.memory.signals.coordinator import MemoryBatchContext, MemorySignalCoordinator +from deerflow.agents.memory.signals.openai_decisions import OpenAIDecisionsSignalClassifier +from deerflow.agents.memory.signals.typesafe import QUESTION_AFFIRMATION, QUESTION_NEGATION, TypeSafeSignalClassifier +from deerflow.guardrails.openai_decisions import OpenAIDecisionsGuardrailError, OpenAIDecisionsGuardrailProvider +from deerflow.guardrails.provider import GuardrailRequest +from deerflow.openai_decisions.client import ( + CATEGORY_REFUSAL, + DEFAULT_API_KEY_ENV, + DEFAULT_BASE_URL, + ENDPOINT_PATH, + QUESTION_CHOICE, + QUESTION_PREDICATE, + OpenAIDecisionsClient, + resolve_decisions_connection, +) +from deerflow.typesafe.client import ( + CATEGORY_LABEL, + CATEGORY_MISSING, + CATEGORY_PROBABILITY, + CATEGORY_TYPE, + MAX_RESPONSE_BYTES, + QUESTION_NOUL, + NoulAnswer, + Question, +) +from deerflow.typesafe.errors import CAUSE_HTTP_STATUS, CAUSE_INVALID_RESPONSE, TypeSafeError + +_API_KEY = "decisions-test-key" +_MODEL = "gpt-decisions-test" +_RUBRIC_TRUE = "it deletes data" +_RUBRIC_FALSE = "it only reads" +_PREDICATE = Question(type=QUESTION_NOUL, instructions="Is this risky?", criteria={"true": _RUBRIC_TRUE, "false": _RUBRIC_FALSE}) +_CHOICE = Question(type=QUESTION_CHOICE, instructions="Which label?", criteria={"keep": "worth keeping", "drop": "not worth keeping"}) + +#: The canonical state JSON the gate digests, written out by hand rather than re-derived by the +#: code under test: the assertion below would pass even if the serialisation changed shape. +_EXPECTED_GATE_INPUT = '{"tool_call":{"arguments":"{\\"command\\":\\"ls\\"}","name":"bash"}}' + + +class _Server: + """Fake Decisions endpoint that records every request it receives.""" + + def __init__(self, responder=None) -> None: + self.requests: list[httpx.Request] = [] + self._responder = responder or (lambda request: httpx.Response(200, json={"answers": [_predicate_answer("q", 0.9)]})) + + def handle(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + return self._responder(request) + + def transport(self) -> httpx.MockTransport: + return httpx.MockTransport(self.handle) + + @property + def body(self) -> dict: + return json.loads(self.requests[-1].content.decode("utf-8")) + + +def _predicate_probability_body(name: str, probability_json: str) -> bytes: + return f'{{"answers":[{{"type":"{QUESTION_PREDICATE}","name":"{name}","probability":{probability_json}}}]}}'.encode() + + +def _predicate_answer(name: str, probability: float) -> dict: + return {"type": QUESTION_PREDICATE, "name": name, "probability": probability} + + +def _client(server: _Server, **settings) -> OpenAIDecisionsClient: + connection = resolve_decisions_connection(settings={"api_key": _API_KEY, "model": _MODEL, **settings}, configuration_source="tests.openai_decisions") + return OpenAIDecisionsClient(connection, transport_factory=server.transport) + + +def _ask(server: _Server, questions: dict[str, Question], input_value: object = "judged text") -> object: + return _client(server).ask(input_value, questions) + + +# --- request mapping -------------------------------------------------------- + + +class TestRequestMapping: + def test_a_noul_question_leaves_as_a_predicate_question(self): + server = _Server() + + _ask(server, {"q": _PREDICATE}) + + question = server.body["questions"][0] + assert question["type"] == QUESTION_PREDICATE + assert question["name"] == "q" + + def test_the_rubric_travels_because_a_predicate_question_has_no_field_for_it(self): + """Dropping the criteria would move the decision boundary while the adapter kept reporting + the same ``criteria`` in its policy identity.""" + server = _Server() + + _ask(server, {"q": _PREDICATE}) + + instructions = server.body["questions"][0]["instructions"] + assert instructions.startswith("Is this risky?") + assert _RUBRIC_TRUE in instructions + assert _RUBRIC_FALSE in instructions + + def test_a_choice_question_keeps_its_instructions_and_carries_its_options(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": QUESTION_CHOICE, "name": "q", "choice": "keep", "confidence": 0.9}]})) + + _ask(server, {"q": _CHOICE}) + + question = server.body["questions"][0] + assert question == { + "type": QUESTION_CHOICE, + "name": "q", + "instructions": "Which label?", + "choices": [{"value": "keep", "description": "worth keeping"}, {"value": "drop", "description": "not worth keeping"}], + } + + def test_the_judged_value_is_input_and_no_state_envelope_is_sent(self): + server = _Server() + + _ask(server, {"q": _PREDICATE}, input_value="the batch text") + + assert server.body["input"] == "the batch text" + assert "state" not in server.body + assert server.body["model"] == _MODEL + + def test_it_posts_to_the_decisions_endpoint_with_the_shared_headers(self): + server = _Server() + + _ask(server, {"q": _PREDICATE}) + + request = server.requests[-1] + assert request.url.path == ENDPOINT_PATH + assert request.method == "POST" + assert request.headers["authorization"] == f"Bearer {_API_KEY}" + # The bounded body read depends on this: an encoded body would make the *decoded* size the + # number that matters, and the client would be counting the wrong thing. + assert request.headers["accept-encoding"] == "identity" + + def test_questions_keep_their_request_order(self): + server = _Server() + + _ask(server, {QUESTION_NEGATION: _PREDICATE, QUESTION_AFFIRMATION: _PREDICATE}) + + assert [entry["name"] for entry in server.body["questions"]] == [QUESTION_NEGATION, QUESTION_AFFIRMATION] + + +class TestConnectionDefaults: + @pytest.mark.parametrize("provider", [OpenAIDecisionsGuardrailProvider, OpenAIDecisionsMemoryPrescreen, OpenAIDecisionsSignalClassifier]) + @pytest.mark.parametrize("settings, field", [({"api_key": "", "model": _MODEL}, "API key"), ({"api_key": _API_KEY}, "model")]) + def test_construction_errors_name_the_configured_service(self, provider, settings, field): + with pytest.raises(ValueError) as excinfo: + provider(**settings) + + message = str(excinfo.value) + assert "OpenAI Decisions" in message + assert "TypeSafe" not in message + assert field in message + assert _API_KEY not in message + + def test_the_backend_defaults_to_its_own_service_not_the_jev_one(self): + server = _Server() + + client = _client(server) + + assert client.connection.base_url == DEFAULT_BASE_URL + assert client.connection.url == f"{DEFAULT_BASE_URL}{ENDPOINT_PATH}" + assert client.connection.api_key_env == DEFAULT_API_KEY_ENV + + def test_a_deployment_must_name_its_model_rather_than_inheriting_another_services_default(self): + """No checked-in model id: an empty default fails loudly at construction.""" + with pytest.raises(ValueError, match="model must be a non-empty string"): + resolve_decisions_connection(settings={"api_key": _API_KEY}, configuration_source="tests.openai_decisions") + + def test_an_operator_override_still_wins(self): + server = _Server() + + client = _client(server, base_url="http://127.0.0.1:8000", model="local-decisions") + + assert client.connection.url == "http://127.0.0.1:8000/v1/decisions" + + +# --- response mapping ------------------------------------------------------ + + +class TestResponseMapping: + def test_a_probability_comes_back_as_the_shared_answer_type(self): + server = _Server(lambda request: httpx.Response(200, json={"model": "served-1.2", "answers": [_predicate_answer("q", 0.31)]})) + + answers = _ask(server, {"q": _PREDICATE}) + + assert answers.answers["q"] == NoulAnswer(probability=0.31) + assert answers.model == "served-1.2" + + def test_a_response_without_a_model_records_the_one_that_was_asked_for(self): + """Provenance is still needed for the audit trail, and the requested model is the only + value here that the endpoint did not choose.""" + server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("q", 0.31)]})) + + answers = _ask(server, {"q": _PREDICATE}) + + assert answers.model == _MODEL + + def test_a_refusal_is_a_per_question_outcome_not_a_request_failure(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": "refusal", "name": "q"}]})) + + answers = _ask(server, {"q": _PREDICATE}) + + assert answers.answers == {} + error = answers.errors_by_question["q"] + assert error.category == CATEGORY_REFUSAL + assert "refused to answer" in error.message + + def test_a_valid_answer_survives_an_invalid_one_in_the_same_array(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("good", 0.4), {"type": QUESTION_PREDICATE, "name": "bad", "probability": "0.9"}]})) + + answers = _ask(server, {"good": _PREDICATE, "bad": _PREDICATE}) + + assert answers.answers["good"] == NoulAnswer(probability=0.4) + assert answers.errors_by_question["bad"].category == CATEGORY_PROBABILITY + + @pytest.mark.parametrize( + ("probability_json", "expected"), + [ + ("true", "non-numeric"), # bool is an int subclass, so True would judge as 1.0 + ('"0.9"', "non-numeric"), # a JSON string must not silently become a number + ("1.5", "outside [0, 1]"), + ("-0.1", "outside [0, 1]"), + ("Infinity", "outside [0, 1]"), # a bare literal json.loads accepts and a threshold cannot + ], + ) + def test_the_shared_numeric_rules_apply_to_this_dialect_too(self, probability_json, expected): + """Written as raw JSON text rather than a Python value, because the payload under test is + what an endpoint can actually put on the wire.""" + server = _Server(lambda request: httpx.Response(200, content=_predicate_probability_body("q", probability_json), headers={"content-type": "application/json"})) + + answers = _ask(server, {"q": _PREDICATE}) + + error = answers.errors_by_question["q"] + assert error.category == CATEGORY_PROBABILITY + assert expected in error.message + assert probability_json not in error.message, "the response body is never repeated into a message" + + def test_a_probability_answered_to_the_wrong_question_kind_is_not_a_verdict(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": "score", "name": "q", "score": 3}]})) + + answers = _ask(server, {"q": _PREDICATE}) + + assert answers.errors_by_question["q"].category == CATEGORY_TYPE + + def test_a_choice_label_outside_the_asked_set_is_rejected(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": QUESTION_CHOICE, "name": "q", "choice": "shred"}]})) + + answers = _ask(server, {"q": _CHOICE}) + + assert answers.errors_by_question["q"].category == CATEGORY_LABEL + + def test_the_first_answer_for_a_repeated_name_wins(self): + """An array can repeat a name where a mapping cannot; which of two conflicting answers a + consumer acts on must not depend on server ordering.""" + server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("q", 0.2), _predicate_answer("q", 0.9)]})) + + answers = _ask(server, {"q": _PREDICATE}) + + assert answers.answers["q"] == NoulAnswer(probability=0.2) + + def test_an_answer_for_a_question_that_was_never_asked_is_ignored_not_reported(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("q", 0.2), _predicate_answer("somebody_elses_question", 0.95)]})) + + answers = _ask(server, {"q": _PREDICATE}) + + assert set(answers.answers) == {"q"} + assert answers.errors_by_question == {} + + def test_a_missing_entry_is_reported_as_missing(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": []})) + + answers = _ask(server, {"q": _PREDICATE}) + + assert answers.errors_by_question["q"].category == CATEGORY_MISSING + + +class TestEnvelopeFailures: + def test_a_non_200_names_this_service_and_not_the_other_one(self): + server = _Server(lambda request: httpx.Response(500)) + + with pytest.raises(TypeSafeError) as excinfo: + _ask(server, {"q": _PREDICATE}) + + assert excinfo.value.cause == CAUSE_HTTP_STATUS + assert "OpenAI Decisions returned HTTP 500" in str(excinfo.value) + + def test_an_answers_object_is_not_an_array_and_is_not_a_verdict(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": {"q": {"probability": 0.9}}})) + + with pytest.raises(TypeSafeError, match="'answers' array") as excinfo: + _ask(server, {"q": _PREDICATE}) + + assert excinfo.value.cause == CAUSE_INVALID_RESPONSE + + def test_an_oversized_body_is_refused_rather_than_read(self): + filler = "x" * (MAX_RESPONSE_BYTES + 1024) + server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("q", 0.5), {"type": "refusal", "name": filler}]})) + + with pytest.raises(TypeSafeError, match="exceeded") as excinfo: + _ask(server, {"q": _PREDICATE}) + + assert excinfo.value.cause == CAUSE_INVALID_RESPONSE + + def test_a_compressed_body_is_refused_rather_than_expanded(self): + """Refused by its encoding, not by a parse failure: this body decodes to a valid envelope.""" + envelope = json.dumps({"answers": [_predicate_answer("q", 0.5)]}).encode() + server = _Server(lambda request: httpx.Response(200, content=gzip.compress(envelope), headers={"Content-Encoding": "gzip"})) + + with pytest.raises(TypeSafeError, match="non-identity") as excinfo: + _ask(server, {"q": _PREDICATE}) + + assert excinfo.value.cause == CAUSE_INVALID_RESPONSE + + +# --- the tool gate --------------------------------------------------------- + + +def _gate(server: _Server, **kwargs) -> OpenAIDecisionsGuardrailProvider: + return OpenAIDecisionsGuardrailProvider(api_key=_API_KEY, model=_MODEL, transport_factory=server.transport, **kwargs) + + +def _tool_call() -> GuardrailRequest: + return GuardrailRequest(tool_name="bash", tool_input={"command": "ls"}) + + +class TestGuardrailGate: + def test_a_probability_at_the_threshold_denies_and_names_this_provider(self): + server = _Server(lambda request: httpx.Response(200, json={"model": "served-1.2", "answers": [_predicate_answer("risky_tool_call", 0.5)]})) + + decision = _gate(server, threshold=0.5).evaluate(_tool_call()) + + assert decision.allow is False + assert decision.reasons[0].code == "openai_decisions.tool_call_risky" + assert "model=served-1.2" in decision.reasons[0].message + + def test_a_probability_below_the_threshold_allows(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("risky_tool_call", 0.4999)]})) + + decision = _gate(server, threshold=0.5).evaluate(_tool_call()) + + assert decision.allow is True + assert decision.reasons[0].code == "openai_decisions.allowed" + + def test_the_bytes_judged_are_the_bytes_the_audit_digest_identifies(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer("risky_tool_call", 0.9)]})) + + decision = _gate(server).evaluate(_tool_call()) + + sent = server.body["input"] + assert sent == _EXPECTED_GATE_INPUT + assert decision.metadata["state_digest"] == hashlib.sha256(sent.encode("utf-8")).hexdigest()[:16] + + def test_a_refusal_denies_for_want_of_a_verdict(self): + """The gate has no verdict without its answer, so a refusal is a denial here while the + same outcome is 'no opinion' for the memory sides.""" + server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": "refusal", "name": "risky_tool_call"}]})) + + with pytest.raises(OpenAIDecisionsGuardrailError) as excinfo: + _gate(server).evaluate(_tool_call()) + + assert excinfo.value.cause == CAUSE_INVALID_RESPONSE + assert "refused to answer" in str(excinfo.value) + + def test_a_request_failure_is_the_provider_error_family_the_middleware_already_maps(self): + server = _Server(lambda request: httpx.Response(503)) + + with pytest.raises(OpenAIDecisionsGuardrailError): + _gate(server).evaluate(_tool_call()) + + def test_an_oversized_call_is_denied_locally_without_a_request(self): + server = _Server() + + decision = _gate(server, max_state_chars=10).evaluate(GuardrailRequest(tool_name="bash", tool_input={"command": "rm -rf /tmp/anything-at-all"})) + + assert decision.allow is False + assert decision.reasons[0].code == "openai_decisions.state_unusable" + assert server.requests == [] + + +# --- the memory sides ------------------------------------------------------ + + +def _prescreen(server: _Server, **kwargs) -> OpenAIDecisionsMemoryPrescreen: + return OpenAIDecisionsMemoryPrescreen(mode=MODE_ENFORCE, api_key=_API_KEY, model=_MODEL, transport_factory=server.transport, **kwargs) + + +class TestMemorySides: + def test_the_pre_screen_skips_below_its_threshold_on_the_same_text_it_always_judged(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": [_predicate_answer(PRESCREEN_QUESTION, 0.1)]})) + provider = _prescreen(server, skip_threshold=0.2) + + decision = provider.decide(_prescreen_request("user: keep using uv\nassistant: noted")) + + assert decision.verdict == "skip" + assert server.body["input"] == "user: keep using uv\nassistant: noted" + assert server.body["questions"][0]["name"] == PRESCREEN_QUESTION + + def test_a_refusal_is_no_opinion_so_the_batch_still_extracts(self): + server = _Server(lambda request: httpx.Response(200, json={"answers": [{"type": "refusal", "name": PRESCREEN_QUESTION}]})) + + assert _prescreen(server).decide(_prescreen_request("anything")) is None + + def test_both_directions_travel_in_one_request_and_a_refusal_only_costs_its_own_label(self): + server = _Server( + lambda request: httpx.Response( + 200, + json={"answers": [_predicate_answer(QUESTION_AFFIRMATION, 0.8), {"type": "refusal", "name": QUESTION_NEGATION}]}, + ) + ) + provider = OpenAIDecisionsSignalClassifier(mode="hints", api_key=_API_KEY, model=_MODEL, transport_factory=server.transport) + + decision = provider.decide(_signal_request("user: yes keep using uv")) + + assert [entry["name"] for entry in server.body["questions"]] == [QUESTION_AFFIRMATION, QUESTION_NEGATION] + assert decision.probabilities == {LABEL_REINFORCEMENT: 0.8} + assert decision.labels == frozenset({LABEL_REINFORCEMENT}) + + def test_the_two_backends_never_share_one_combined_request(self): + """``sharing_key`` is what lets two sides travel together; a different endpoint or a + different credential must separate them or a verdict could be served to the wrong service.""" + server = _Server() + settings = {"api_key": _API_KEY, "model": _MODEL, "base_url": "https://decision-gateway.example", "transport_factory": server.transport} + left = TypeSafeMemoryPrescreen(**settings) + right = OpenAIDecisionsMemoryPrescreen(**settings) + + assert left._connection.url != right._connection.url + assert left.sharing_key() != right.sharing_key() + + def test_it_resolves_from_the_existing_provider_slot_without_a_schema_change(self): + provider = resolve_memory_prescreen( + mode="shadow", + use="deerflow.agents.memory.prescreen.openai_decisions:OpenAIDecisionsMemoryPrescreen", + config={"api_key": _API_KEY, "model": _MODEL}, + ) + + assert provider.name == "openai_decisions" + assert provider.policy_id == "deerflow.memory.prescreen.openai_decisions" + + +def _gateway_response(request: httpx.Request) -> httpx.Response: + """One gateway serves both dialects, with different classifier opinions.""" + questions = json.loads(request.content)["questions"] + if request.url.path == "/v1/systemone": + assert isinstance(questions, dict) + return httpx.Response(200, json={"model": "jev-test", "answers": {name: {"type": "noul", "noul": 0.1} for name in questions}}) + assert request.url.path == "/v1/decisions" + probabilities = {PRESCREEN_QUESTION: 0.1, QUESTION_AFFIRMATION: 0.9, QUESTION_NEGATION: 0.1} + return httpx.Response(200, json={"answers": [_predicate_answer(question["name"], probabilities[question["name"]]) for question in questions]}) + + +def _judge_gateway(prescreen_type, classifier_type, combine): + server = _Server(_gateway_response) + settings = {"api_key": _API_KEY, "model": _MODEL, "base_url": "https://decision-gateway.example", "transport_factory": server.transport} + coordinator = MemorySignalCoordinator( + prescreen=prescreen_type(mode="enforce", **settings), + prescreen_mode="enforce", + classifier=classifier_type(mode="hints", **settings), + classifier_mode="hints", + combine=combine, + ) + text = "user: Continue using the same editor preference." + verdict = coordinator.judge(MemoryBatchContext(batch_text=text, digest=hashlib.sha256(text.encode()).hexdigest()[:16])) + return server, verdict + + +class TestSameGatewayCoordination: + @pytest.mark.parametrize("combine", ["auto", "never"]) + @pytest.mark.parametrize( + "prescreen_type, classifier_type, paths, veto", + [ + (TypeSafeMemoryPrescreen, OpenAIDecisionsSignalClassifier, ["/v1/systemone", "/v1/decisions"], True), + (OpenAIDecisionsMemoryPrescreen, TypeSafeSignalClassifier, ["/v1/decisions", "/v1/systemone"], False), + ], + ) + def test_mixed_backends_keep_their_requests_and_verdicts(self, combine, prescreen_type, classifier_type, paths, veto): + server, verdict = _judge_gateway(prescreen_type, classifier_type, combine) + + assert [request.url.path for request in server.requests] == paths + for request, expected_questions in zip(server.requests, [{PRESCREEN_QUESTION}, {QUESTION_AFFIRMATION, QUESTION_NEGATION}], strict=True): + body = json.loads(request.content) + if request.url.path == "/v1/systemone": + assert set(body["questions"]) == expected_questions + assert "state" in body and "input" not in body + else: + assert {question["name"] for question in body["questions"]} == expected_questions + assert "input" in body and "state" not in body + assert verdict.skip is not veto + assert verdict.vetoed_by_model_signal is veto + assert verdict.hints == (frozenset({LABEL_REINFORCEMENT}) if veto else frozenset()) + + @pytest.mark.parametrize( + "prescreen_type, classifier_type, path, veto", + [ + (TypeSafeMemoryPrescreen, TypeSafeSignalClassifier, "/v1/systemone", False), + (OpenAIDecisionsMemoryPrescreen, OpenAIDecisionsSignalClassifier, "/v1/decisions", True), + ], + ) + def test_matching_backends_still_combine(self, prescreen_type, classifier_type, path, veto): + server, verdict = _judge_gateway(prescreen_type, classifier_type, "auto") + + assert [request.url.path for request in server.requests] == [path] + questions = json.loads(server.requests[0].content)["questions"] + names = set(questions) if path == "/v1/systemone" else {question["name"] for question in questions} + assert names == {PRESCREEN_QUESTION, QUESTION_AFFIRMATION, QUESTION_NEGATION} + assert verdict.skip is not veto + assert verdict.vetoed_by_model_signal is veto + + +def _prescreen_request(text: str): + from deerflow.agents.memory.prescreen.contract import MemoryPrescreenRequest + + return MemoryPrescreenRequest(batch_text=text, digest=hashlib.sha256(text.encode()).hexdigest()[:16]) + + +def _signal_request(text: str): + from deerflow.agents.memory.signals.contract import MemorySignalRequest + + return MemorySignalRequest(batch_text=text, digest=hashlib.sha256(text.encode()).hexdigest()[:16]) diff --git a/backend/tests/test_typesafe_config.py b/backend/tests/test_typesafe_config.py index 3e4ac89c93..c71cf67ecd 100644 --- a/backend/tests/test_typesafe_config.py +++ b/backend/tests/test_typesafe_config.py @@ -192,6 +192,15 @@ class TestModeOff: class TestIdentities: + def test_endpoint_path_separates_sharing_and_public_policy_identity(self): + settings = {"api_key": "key-a", "model": "gateway-model", "base_url": "https://gateway.example"} + jev = resolve_connection(settings=settings, configuration_source="tests.typesafe") + decisions = resolve_connection(settings=settings, configuration_source="tests.decisions", endpoint_path="/v1/decisions") + + assert TypeSafeClient(jev).sharing_key() != TypeSafeClient(decisions).sharing_key() + assert jev.public_parameters()["endpoint_path"] == "/v1/systemone" + assert decisions.public_parameters()["endpoint_path"] == "/v1/decisions" + def test_the_sharing_key_tracks_every_sharing_dimension(self): base = {"api_key": "key-a", "model": "jev-latest"} baseline = _client(**base).sharing_key(max_state_chars=100) diff --git a/config.example.yaml b/config.example.yaml index 727b5b4d67..1039ef8204 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -3295,6 +3295,24 @@ run_ownership: # # max_state_chars / cache_size / cache_ttl_seconds / # # instructions / criteria +# --- Option 4b: the same risk gate over OpenAI Decisions ------------------- +# Same provider semantics, another endpoint: one `predicate` question about the +# call to POST /v1/decisions, denied when its probability reaches `threshold`. +# Nothing in the gate is re-decided -- the strict-JSON local preflight, the +# character-count limit, the cache and fail_closed are the Option 4 ones, and +# every other Option 4 setting below applies here unchanged. +# guardrails: +# enabled: true +# fail_closed: true +# provider: +# use: deerflow.guardrails.openai_decisions:OpenAIDecisionsGuardrailProvider +# config: +# model: # required: this +# # backend ships no default model, so batches cannot go to a model nobody chose +# api_key_env: OPENAI_API_KEY # default; base_url defaults to https://api.openai.com +# threshold: 0.5 +# tools: ["bash"] + # ============================================================================ # TypeSafe (Jev) Client Defaults (optional) # ============================================================================