feat(llm): shared LLM egress (CROSS-1) + model-agnostic adapter (#4) — PR-1#2
Closed
gesh75 wants to merge 2 commits into
Closed
feat(llm): shared LLM egress (CROSS-1) + model-agnostic adapter (#4) — PR-1#2gesh75 wants to merge 2 commits into
gesh75 wants to merge 2 commits into
Conversation
…[PR-1] One egress point for every LLM call, with a 4-event hook bus so #2 (caching), #4 (model-identity), and #1 (Reflexion) attach without forking the egress — the dependency spine the whole upgrade set rides on. - core/llm/egress.py: LLMEgress (ordered local-first fallback chain, retry/backoff/cooldown = LiteLLM logic, no dep) + HookBus (isolated dispatch: a broken hook never breaks a call) + frozen LLMRequest - core/llm/backends.py: OpenAICompatLocalBackend + AnthropicCloudBackend via RAW httpx (NO `import anthropic` — keeps the air-gap invariant provable) + frozen LLMResult / ModelIdentity / LLMAttempt - core/llm/{errors,airgap,config,identity}.py: typed taxonomy + is_retryable; air-gap fail-closed (is_loopback; refuse cloud/non-loopback before a socket); env-driven immutable config (local-first; cloud dropped in air-gap); local weights-sha256 vs honest cloud identity-claim (never a faked weight hash) - core/llm/adapter.py: LLMAdapter facade delegating to the ONE egress (no 2nd chain) - tests/llm_egress_test.py: 28 tests (taxonomy, is_loopback, air-gap, immutability, wired fallback/retry/cooldown/fail-closed, hook isolation, httpx MockTransport backend over no sockets, single-egress anthropic scan, no-litellm). 28/28 green. Deferred to the cutover PR: the "no /v1/chat/completions outside core/llm" scan (HttpBackend.generate_config still owns its POST). All 6 existing suites stay green.
Reviewer's GuideIntroduces a new, single shared LLM egress pipeline with a local-first fallback chain, typed error taxonomy, air‑gap enforcement, model identity resolution, and a thin adapter façade, plus tests that gate the new behavior and invariants without touching existing files. Sequence diagram for shared LLM egress completion flowsequenceDiagram
actor Caller
participant Adapter as LLMAdapter
participant Egress as LLMEgress
participant Hooks as HookBus
participant Backend as LLMBackend
Caller->>Adapter: complete_sync(system, user,...)
Adapter->>Egress: complete_sync(LLMRequest)
alt no running event loop
Egress->>Egress: complete(req)
else event loop running
Egress->>Egress: complete(req) in worker thread
end
Egress->>Hooks: transform(before_request, req)
Hooks-->>Egress: LLMRequest
loop ordered_backends(tier)
Egress->>Egress: _is_cooled_down(provider)?
alt not cooled down
Egress->>Backend: complete(system, user, max_tokens, temperature)
alt success
Backend-->>Egress: text, ModelIdentity
Egress->>Hooks: observe(after_response, provider, identity)
Egress->>Hooks: observe(on_result, LLMResult)
Egress-->>Adapter: LLMResult
Adapter-->>Caller: LLMResult
Egress->>Egress: break
else AdapterError
Egress->>Egress: is_retryable(err)?
alt retryable
Egress->>Egress: asyncio.sleep(_backoff_s)
else not retryable
Egress->>Egress: _cool_down(provider) (if RateLimited)
Egress->>Egress: try next backend
end
end
end
end
Egress-->>Caller: AdapterError (all backends failed)
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 3 issues, and left some high level feedback:
- LLMEgress.complete_sync spins up a new ThreadPoolExecutor and nested asyncio.run for every call; consider reusing an executor or using asyncio.to_thread/loop.run_in_executor to avoid the per-call thread pool creation and nested event loop overhead.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- LLMEgress.complete_sync spins up a new ThreadPoolExecutor and nested asyncio.run for every call; consider reusing an executor or using asyncio.to_thread/loop.run_in_executor to avoid the per-call thread pool creation and nested event loop overhead.
## Individual Comments
### Comment 1
<location path="core/llm/backends.py" line_range="113-122" />
<code_context>
+ )
+ raise err
+
+ try:
+ text = resp.json()["content"][0]["text"]
+ except (KeyError, IndexError, ValueError) as exc:
+ raise Permanent(
</code_context>
<issue_to_address>
**issue (bug_risk):** Response parsing for Anthropic messages is brittle and assumes a simplified schema that may not match actual API responses.
The `/v1/messages` response uses a `content` list of typed blocks (e.g., `{"type": "text", "text": "..."}`). Accessing `resp.json()["content"][0]["text"]` will fail if the first block is non-text (e.g., tool calls) or if keys/indices differ slightly, causing valid responses to be treated as `Permanent`. Consider iterating `content` to find the first `type == "text"` block, guarding missing keys more defensively, and deciding how to handle multi-block content (e.g., concatenate text segments).
</issue_to_address>
### Comment 2
<location path="core/llm/backends.py" line_range="97" />
<code_context>
+ """Build config + backends from env (air-gap checks fire at backend construction)."""
+ return cls(LLMEgress.from_env(hooks=hooks))
+
+ async def complete(
+ self,
+ *,
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the shared HTTP/error-handling and JSON-parsing logic into helpers and caching model identity per backend instance to simplify these backend implementations.
You can reduce duplication and make the control flow easier to follow by extracting the shared HTTP + error-handling and JSON parsing skeletons into small helpers, and by caching identity per backend instance.
### 1) Extract shared HTTP + classification logic
Both `complete` implementations share:
- `AsyncClient` POST
- `httpx.RequestError` → `classify(None, ...)`
- `status_code >= 400` → `classify(...)` + `RateLimited` header patching
You can centralize that without changing behavior:
```python
# keep in this module, near _retry_after_seconds
from .errors import RateLimited
async def _post_with_classified_errors(
*,
url: str,
body: dict[str, object],
provider: Provider,
timeout_s: float,
transport: httpx.BaseTransport | None,
headers: dict[str, str] | None = None,
) -> httpx.Response:
try:
async with httpx.AsyncClient(transport=transport, timeout=timeout_s) as client:
resp = await client.post(url, json=body, headers=headers)
except httpx.RequestError as exc:
raise classify(None, str(exc), provider) from exc
if resp.status_code >= 400:
err = classify(resp.status_code, resp.text, provider)
if isinstance(err, RateLimited) and err.retry_after_s is None:
hdr = _retry_after_seconds(resp)
if hdr is not None:
err = RateLimited(
message=err.message,
provider=err.provider,
status=err.status,
retry_after_s=hdr,
)
raise err
return resp
```
Then each backend’s `complete` becomes shorter and the subtle rate-limit logic is centralized:
```python
# OpenAICompatLocalBackend.complete
url = f"{self._spec.base_url}/v1/chat/completions"
body = {
"model": self._spec.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"temperature": temperature,
"stream": False,
"max_tokens": max_tokens,
}
resp = await _post_with_classified_errors(
url=url,
body=body,
provider=self.provider,
timeout_s=self._timeout_s,
transport=self._transport,
)
# AnthropicCloudBackend.complete
url = f"{self._spec.base_url}/v1/messages"
body = {
"model": self._spec.model,
"max_tokens": max_tokens,
"temperature": temperature,
"system": system,
"messages": [{"role": "user", "content": user}],
}
resp = await _post_with_classified_errors(
url=url,
body=body,
headers=headers,
provider=self.provider,
timeout_s=self._timeout_s,
transport=self._transport,
)
```
### 2) Extract success-path JSON parsing skeleton
The success path is also structurally identical aside from the JSON path and error message. A small helper that wraps `resp.json()` + `Permanent` preserves behavior and removes duplicated try/excepts:
```python
from collections.abc import Callable
from .errors import Permanent
def _extract_text(
resp: httpx.Response,
*,
provider: Provider,
parse: Callable[[object], str],
malformed_message: str,
) -> str:
try:
payload = resp.json()
return parse(payload)
except (KeyError, IndexError, ValueError, TypeError) as exc:
raise Permanent(
message=malformed_message,
provider=provider,
status=resp.status_code,
) from exc
```
Usage:
```python
# OpenAICompatLocalBackend.complete
text = _extract_text(
resp,
provider=self.provider,
malformed_message="malformed OpenAI-compat completion response",
parse=lambda payload: payload["choices"][0]["message"]["content"],
)
# AnthropicCloudBackend.complete
text = _extract_text(
resp,
provider=self.provider,
malformed_message="malformed Anthropic messages response",
parse=lambda payload: payload["content"][0]["text"],
)
return text, self.identity()
```
This keeps the provider-specific JSON paths explicit but moves the defensive scaffolding into one place.
### 3) Cache identity per backend instance
Given `BackendSpec` is immutable, `ModelIdentity` is effectively constant per backend. You can compute it once and reuse it, avoiding repeated hashing/lookup and clarifying semantics:
```python
class OpenAICompatLocalBackend:
provider: Provider = "openai-compatible-local"
def __init__(..., spec: BackendSpec, ...):
...
from .identity import resolve_model_identity
self._identity = resolve_model_identity(spec)
def identity(self) -> ModelIdentity:
return self._identity
```
Same for `AnthropicCloudBackend`. This change is internal and preserves the public API while making `complete`’s cost and lifecycle easier to reason about.
</issue_to_address>
### Comment 3
<location path="core/llm/egress.py" line_range="80" />
<code_context>
+ _log_hook_error(event, fn, exc)
+
+
+class LLMEgress:
+ """The single egress. Ordered fallback chain + retry/cooldown + hook bus."""
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider adding typed hook registration helpers, extracting the sync bridge runner, and factoring retry bookkeeping into helpers to make LLMEgress easier to read and extend without changing behavior.
You can keep all current behavior while tightening the surface and simplifying a few dense paths.
### 1. Make hook usage more explicit (typed helpers on `LLMEgress`)
You can keep `HookBus` as-is internally, but give callers typed, self-documenting helpers instead of the generic `on(event, fn)`:
```python
class LLMEgress:
# ...
def on_before_request(self, fn: Callable[[LLMRequest], LLMRequest | None]) -> "LLMEgress":
self._hooks.register("before_request", fn)
return self
def on_after_response(
self,
fn: Callable[[str, ModelIdentity], None],
) -> "LLMEgress":
self._hooks.register("after_response", fn)
return self
def on_result(self, fn: Callable[[LLMResult], None]) -> "LLMEgress":
self._hooks.register("on_result", fn)
return self
# keep the existing generic method for backwards compatibility
def on(self, event: HookEvent, fn: Callable) -> "LLMEgress":
self._hooks.register(event, fn)
return self
```
Call sites for #1/#2/#4 can move to the explicit methods over time, and you still support `build_payload` / new events via `on()` when needed.
### 2. Extract the `asyncio.run` call in `complete_sync`
You can reduce the nested control-flow and remove the inline lambda by extracting the “run the coroutine to completion” piece:
```python
class LLMEgress:
# ...
def _run_complete(self, req: LLMRequest) -> LLMResult:
return asyncio.run(self.complete(req))
def complete_sync(self, req: LLMRequest) -> LLMResult:
try:
asyncio.get_running_loop()
except RuntimeError:
return self._run_complete(req)
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
return pool.submit(self._run_complete, req).result()
```
This keeps the same semantics but makes the path easier to read and test.
### 3. Split retry bookkeeping in `_try_backend`
You can factor out the repeated `LLMAttempt` construction and retry decision without changing behavior:
```python
class LLMEgress:
# ...
def _record_attempt(
self,
attempts: list[LLMAttempt],
backend: LLMBackend,
*,
ok: bool,
model: str,
status: int | None,
error_kind: str | None,
start_t: float,
) -> None:
attempts.append(
LLMAttempt(
provider=backend.provider,
model=model,
ok=ok,
status=status,
error_kind=error_kind,
latency_ms=int((time.perf_counter() - start_t) * 1000),
)
)
def _should_retry(self, err: AdapterError, attempt: int) -> bool:
if not is_retryable(err):
return False
return attempt < self._cfg.max_retries
async def _try_backend(
self, backend: LLMBackend, req: LLMRequest, attempts: list[LLMAttempt]
) -> tuple[str, ModelIdentity] | None:
for attempt in range(self._cfg.max_retries + 1):
t0 = time.perf_counter()
try:
text, identity = await backend.complete(
system=req.system,
user=req.user,
max_tokens=req.max_tokens,
temperature=req.temperature,
)
self._record_attempt(
attempts,
backend,
ok=True,
model=identity.model,
status=200,
error_kind=None,
start_t=t0,
)
return text, identity
except AdapterError as err:
self._record_attempt(
attempts,
backend,
ok=False,
model="unknown",
status=err.status,
error_kind=type(err).__name__,
start_t=t0,
)
if not self._should_retry(err, attempt):
if isinstance(err, RateLimited):
self._cool_down(backend.provider)
return None
await asyncio.sleep(self._backoff_s(attempt, err))
return None
```
This keeps retry/cooldown semantics identical while making the core loop less dense and easier to reason about.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+113
to
+122
| try: | ||
| return float(raw) | ||
| except ValueError: | ||
| return None | ||
|
|
||
|
|
||
| class OpenAICompatLocalBackend: | ||
| """POST {base}/v1/chat/completions — the universal local contract. | ||
|
|
||
| Construction runs the air-gap pre-check (loopback enforced when AEGIS_AIRGAP=1). |
There was a problem hiding this comment.
issue (bug_risk): Response parsing for Anthropic messages is brittle and assumes a simplified schema that may not match actual API responses.
The /v1/messages response uses a content list of typed blocks (e.g., {"type": "text", "text": "..."}). Accessing resp.json()["content"][0]["text"] will fail if the first block is non-text (e.g., tool calls) or if keys/indices differ slightly, causing valid responses to be treated as Permanent. Consider iterating content to find the first type == "text" block, guarding missing keys more defensively, and deciding how to handle multi-block content (e.g., concatenate text segments).
|
|
||
| provider: Provider | ||
|
|
||
| async def complete( |
There was a problem hiding this comment.
issue (complexity): Consider extracting the shared HTTP/error-handling and JSON-parsing logic into helpers and caching model identity per backend instance to simplify these backend implementations.
You can reduce duplication and make the control flow easier to follow by extracting the shared HTTP + error-handling and JSON parsing skeletons into small helpers, and by caching identity per backend instance.
1) Extract shared HTTP + classification logic
Both complete implementations share:
AsyncClientPOSThttpx.RequestError→classify(None, ...)status_code >= 400→classify(...)+RateLimitedheader patching
You can centralize that without changing behavior:
# keep in this module, near _retry_after_seconds
from .errors import RateLimited
async def _post_with_classified_errors(
*,
url: str,
body: dict[str, object],
provider: Provider,
timeout_s: float,
transport: httpx.BaseTransport | None,
headers: dict[str, str] | None = None,
) -> httpx.Response:
try:
async with httpx.AsyncClient(transport=transport, timeout=timeout_s) as client:
resp = await client.post(url, json=body, headers=headers)
except httpx.RequestError as exc:
raise classify(None, str(exc), provider) from exc
if resp.status_code >= 400:
err = classify(resp.status_code, resp.text, provider)
if isinstance(err, RateLimited) and err.retry_after_s is None:
hdr = _retry_after_seconds(resp)
if hdr is not None:
err = RateLimited(
message=err.message,
provider=err.provider,
status=err.status,
retry_after_s=hdr,
)
raise err
return respThen each backend’s complete becomes shorter and the subtle rate-limit logic is centralized:
# OpenAICompatLocalBackend.complete
url = f"{self._spec.base_url}/v1/chat/completions"
body = {
"model": self._spec.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"temperature": temperature,
"stream": False,
"max_tokens": max_tokens,
}
resp = await _post_with_classified_errors(
url=url,
body=body,
provider=self.provider,
timeout_s=self._timeout_s,
transport=self._transport,
)
# AnthropicCloudBackend.complete
url = f"{self._spec.base_url}/v1/messages"
body = {
"model": self._spec.model,
"max_tokens": max_tokens,
"temperature": temperature,
"system": system,
"messages": [{"role": "user", "content": user}],
}
resp = await _post_with_classified_errors(
url=url,
body=body,
headers=headers,
provider=self.provider,
timeout_s=self._timeout_s,
transport=self._transport,
)2) Extract success-path JSON parsing skeleton
The success path is also structurally identical aside from the JSON path and error message. A small helper that wraps resp.json() + Permanent preserves behavior and removes duplicated try/excepts:
from collections.abc import Callable
from .errors import Permanent
def _extract_text(
resp: httpx.Response,
*,
provider: Provider,
parse: Callable[[object], str],
malformed_message: str,
) -> str:
try:
payload = resp.json()
return parse(payload)
except (KeyError, IndexError, ValueError, TypeError) as exc:
raise Permanent(
message=malformed_message,
provider=provider,
status=resp.status_code,
) from excUsage:
# OpenAICompatLocalBackend.complete
text = _extract_text(
resp,
provider=self.provider,
malformed_message="malformed OpenAI-compat completion response",
parse=lambda payload: payload["choices"][0]["message"]["content"],
)
# AnthropicCloudBackend.complete
text = _extract_text(
resp,
provider=self.provider,
malformed_message="malformed Anthropic messages response",
parse=lambda payload: payload["content"][0]["text"],
)
return text, self.identity()This keeps the provider-specific JSON paths explicit but moves the defensive scaffolding into one place.
3) Cache identity per backend instance
Given BackendSpec is immutable, ModelIdentity is effectively constant per backend. You can compute it once and reuse it, avoiding repeated hashing/lookup and clarifying semantics:
class OpenAICompatLocalBackend:
provider: Provider = "openai-compatible-local"
def __init__(..., spec: BackendSpec, ...):
...
from .identity import resolve_model_identity
self._identity = resolve_model_identity(spec)
def identity(self) -> ModelIdentity:
return self._identitySame for AnthropicCloudBackend. This change is internal and preserves the public API while making complete’s cost and lifecycle easier to reason about.
| _log_hook_error(event, fn, exc) | ||
|
|
||
|
|
||
| class LLMEgress: |
There was a problem hiding this comment.
issue (complexity): Consider adding typed hook registration helpers, extracting the sync bridge runner, and factoring retry bookkeeping into helpers to make LLMEgress easier to read and extend without changing behavior.
You can keep all current behavior while tightening the surface and simplifying a few dense paths.
1. Make hook usage more explicit (typed helpers on LLMEgress)
You can keep HookBus as-is internally, but give callers typed, self-documenting helpers instead of the generic on(event, fn):
class LLMEgress:
# ...
def on_before_request(self, fn: Callable[[LLMRequest], LLMRequest | None]) -> "LLMEgress":
self._hooks.register("before_request", fn)
return self
def on_after_response(
self,
fn: Callable[[str, ModelIdentity], None],
) -> "LLMEgress":
self._hooks.register("after_response", fn)
return self
def on_result(self, fn: Callable[[LLMResult], None]) -> "LLMEgress":
self._hooks.register("on_result", fn)
return self
# keep the existing generic method for backwards compatibility
def on(self, event: HookEvent, fn: Callable) -> "LLMEgress":
self._hooks.register(event, fn)
return selfCall sites for #1/#2/#4 can move to the explicit methods over time, and you still support build_payload / new events via on() when needed.
2. Extract the asyncio.run call in complete_sync
You can reduce the nested control-flow and remove the inline lambda by extracting the “run the coroutine to completion” piece:
class LLMEgress:
# ...
def _run_complete(self, req: LLMRequest) -> LLMResult:
return asyncio.run(self.complete(req))
def complete_sync(self, req: LLMRequest) -> LLMResult:
try:
asyncio.get_running_loop()
except RuntimeError:
return self._run_complete(req)
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
return pool.submit(self._run_complete, req).result()This keeps the same semantics but makes the path easier to read and test.
3. Split retry bookkeeping in _try_backend
You can factor out the repeated LLMAttempt construction and retry decision without changing behavior:
class LLMEgress:
# ...
def _record_attempt(
self,
attempts: list[LLMAttempt],
backend: LLMBackend,
*,
ok: bool,
model: str,
status: int | None,
error_kind: str | None,
start_t: float,
) -> None:
attempts.append(
LLMAttempt(
provider=backend.provider,
model=model,
ok=ok,
status=status,
error_kind=error_kind,
latency_ms=int((time.perf_counter() - start_t) * 1000),
)
)
def _should_retry(self, err: AdapterError, attempt: int) -> bool:
if not is_retryable(err):
return False
return attempt < self._cfg.max_retries
async def _try_backend(
self, backend: LLMBackend, req: LLMRequest, attempts: list[LLMAttempt]
) -> tuple[str, ModelIdentity] | None:
for attempt in range(self._cfg.max_retries + 1):
t0 = time.perf_counter()
try:
text, identity = await backend.complete(
system=req.system,
user=req.user,
max_tokens=req.max_tokens,
temperature=req.temperature,
)
self._record_attempt(
attempts,
backend,
ok=True,
model=identity.model,
status=200,
error_kind=None,
start_t=t0,
)
return text, identity
except AdapterError as err:
self._record_attempt(
attempts,
backend,
ok=False,
model="unknown",
status=err.status,
error_kind=type(err).__name__,
start_t=t0,
)
if not self._should_retry(err, attempt):
if isinstance(err, RateLimited):
self._cool_down(backend.provider)
return None
await asyncio.sleep(self._backoff_s(attempt, err))
return NoneThis keeps retry/cooldown semantics identical while making the core loop less dense and easier to reason about.
Automated security review (HIGH) flagged the string-prefix host check: a host that
only STARTS WITH "127." (e.g. 127.evil.com) or contains "localhost"
(localhost.evil.com) passed startswith("127.") yet resolves OFF-perimeter — a real
air-gap bypass, since is_loopback gates the entire wedge.
Fix: parse the host with ipaddress.ip_address(...).is_loopback (strict, literal-IP
only) instead of a string prefix. Allow only the literal name localhost/ip6-localhost,
genuine 127.0.0.0/8 + ::1, and the reserved non-routable .local mDNS suffix; reject
empty/relative hosts. NO DNS resolution — resolving an attacker-supplied host would
itself be egress, exactly what air-gap forbids.
Adds test_is_loopback_rejects_spoofed_prefix (127.evil.com / 127.0.0.1.evil.com /
localhost.evil.com / empty all rejected; ::1 + 127.5.5.5 allowed). Suite 29/29 green.
gesh75
force-pushed
the
feat/llm-egress-cross1
branch
from
May 31, 2026 21:08
3b703b4 to
6e71d03
Compare
This was referenced Jun 1, 2026
gesh75
added a commit
that referenced
this pull request
Jun 1, 2026
…authority gate, ceiling, CROSS-3 seal (#9) Squashed release of the 8-PR stack (#2..#8): the offline, hardware-rootable bounded-autonomy receipt and everything under it. AUTO disabled by default; the only un-built piece is the hardware-PIV signer (docs/PIV_HARDWARE_SIGNER_PLAN.md) — software Ed25519 runs the whole demo. - core/llm: single shared LLM egress + adapter (local-first fallback chain, retry/cooldown, 4-event hook bus; raw httpx, no anthropic SDK; air-gap fail-closed). [#2] - evidence: model-identity wedge — change.model_identity sealed into the bundle hash; honest UNKNOWN / unattested states, never a faked attribution. [#3] - core/risk: deterministic AUTO/HITL/HOTL/BLOCK authority model (severity != authority; AS/RD/RT -> BLOCK; spine/underlay -> >= HOTL) + the no-self-escalation ceiling. [#4] - core/promote: gate rule G5 enforces the ceiling from the sealed required tier; change.authority sealed (bundle_version 1.1->1.2). [#5] - core/seal: CROSS-3 detached, offline-verifiable bounded-autonomy receipt (Ed25519; gates G0..G6; refuses to seal an unbounded change). [#6] - serve: /api/preflight/run emits the seal; GET /api/seal/pubkey + POST /api/seal/verify; examiner PDF shows the seal. [#7] - docs: PIV hardware-signer plan — the final, hardware-gated CROSS-3 pass. [#8] Property-tested safety invariants: no-self-escalation 0-violations, every-seal-is-bounded, tamper detection, offline-keyless verify, air-gap fail-closed.
Owner
Author
|
Superseded by the squashed release #9, now merged to main. This PR's changes are in main; closing for tidiness (branch kept for history). |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
PR-1 — the dependency spine
The single LLM egress that #2 (caching), #4 (model-identity), and #1 (Reflexion) all ride. Build it once; later features register a hook instead of forking the egress.
What's here (
aegis/core/llm/, all additive — no existing file touched)egress.py—LLMEgress: ordered local-first fallback chain + retry/exponential-backoff/per-provider cooldown (LiteLLM logic, no dependency).HookBuswith isolated dispatch — a broken hook never breaks an LLM call. 4 events:before_request,build_payload(reserved for feat(llm): shared LLM egress (CROSS-1) + model-agnostic adapter (#4) — PR-1 #2),after_response,on_result.backends.py—OpenAICompatLocalBackend+AnthropicCloudBackendvia raw httpx (noimport anthropic) so the air-gap invariant stays provable. FrozenLLMResult/ModelIdentity/LLMAttempt.errors.py— typed taxonomy (Auth/RateLimited/ContextExceeded/Transient/Permanent) +is_retryable+ Retry-After honoring.airgap.py— fail-closed: refuse cloud / non-loopback before a socket opens; assert SDK never imported.config.py— env-driven immutable config; cloud dropped entirely in air-gap.identity.py— local weights-sha256 attestation (cached) vs honest cloud identity-claim (never a faked weight hash).adapter.py—LLMAdapterfaçade delegating to the one egress (no second chain).Tests —
tests/llm_egress_test.py, 28/28 greentaxonomy ·
is_loopback· air-gap fail-closed · immutability (frozen) · wired fallback/retry/cooldown/fail-closed · hook isolation · backend overhttpx.MockTransport(no sockets) · single-egressanthropic-import scan · no-litellm. All 6 existing AEGIS suites stay green (stress 25k/0, promote 6k/0, twin 8k/0, pdf, contract, api 12/12).Deferred (documented in-test)
The "no
/v1/chat/completionsoutsidecore/llm/" scan —HttpBackend.generate_configstill owns its POST until the cutover PR; enabling it now would be a false RED.Next: #4-wedge (seal
model_identityinto the bundle viaafter_response) → #5-core/ceiling → CROSS-3 SEAL.Summary by Sourcery
Introduce a single shared LLM egress pipeline with a model-agnostic adapter and strict air-gap safeguards, providing unified backend routing, typed error handling, and immutable model identity attestation.
New Features:
Tests: