Self Observation¶
Cluster: Uncategorised | Type: component | MCP Tools: 6
Overview¶
Public API¶
SelfObservationDecisionError(ValueError)¶
The LLM did not produce a usable, validated risk-level decision.
RiskLevelDecision¶
Validated risk-level verdict for a self_observation event classification.
| Field | Type | Default |
|---|---|---|
risk | str | required |
baseline_risk | str | required |
requires_review | bool | False |
degraded | bool | False |
llm_used | bool | False |
anomalies | tuple[str, ...] | () |
rationale | str | '' |
raw_response | str | '' |
Methods:
agentic_evidence() -> dict[str, Any]¶
to_dict() -> dict[str, Any]¶
SelfObservationPlanner¶
Runtime-first facade with deterministic fallback + one-way risk-level clamp.
| Field | Type | Default |
|---|---|---|
runtime | SelfObservationRuntime \| None | None |
last_decision | RiskLevelDecision \| None | field(default=None, init=False) |
last_llm_used | bool | field(default=False, init=False) |
last_fallback_reason | str | field(default='', init=False) |
Methods:
decide(event_or_payload: Any) -> RiskLevelDecision¶
SelfObservationEvent(BaseModel)¶
| Field | Type | Default |
|---|---|---|
event_id | str | Field(default_factory=lambda: uuid.uuid4().hex) |
timestamp | float | Field(default_factory=time.time) |
event_type | str | required |
run_id | str | required |
harness_id | str | required |
task_id | str | '' |
component_id | str | '' |
agent_id | str | '' |
model_id | str | '' |
route_decision | str | '' |
input_summary | str | '' |
output_summary | str | '' |
files_changed | list[str] | Field(default_factory=list) |
commands_run | list[str] | Field(default_factory=list) |
tests_run | list[str] | Field(default_factory=list) |
test_results | dict | Field(default_factory=dict) |
failure_modes_detected | list[str] | Field(default_factory=list) |
failure_modes_present | list[str] | Field(default_factory=list) |
human_interventions | list[str] | Field(default_factory=list) |
cost | float | 0.0 |
latency | float | 0.0 |
tokens | int | 0 |
risk_level | str | 'low' |
rollback_available | bool | False |
docs_updated | bool | False |
source_visualisation_updated | bool | False |
user_feedback | str | '' |
SelfObservationEmitResult(BaseModel)¶
| Field | Type | Default |
|---|---|---|
recorded | bool | required |
event | SelfObservationEvent \| None | None |
reason | str | '' |
SelfObservationBus¶
Constructor:
| Parameter | Type | Default |
|---|---|---|
trace_root | str \| Path | '.g6/traces' |
db_path | str \| Path | 'self_observation_events.db' |
Methods:
emit(event_type: str, payload: dict) -> SelfObservationEvent¶
emit_safe(event_type: str, payload: Mapping[str, Any]) -> SelfObservationEmitResult¶
grounded_emit(event_type: str, payload: Mapping[str, Any], planner: Any = _DEFAULT_PLANNER) -> 'GroundedEmitResult'¶
Classify the event's TRUE risk via the grounded chokepoint, THEN record it.
GroundedEmitResult(BaseModel)¶
Result of a grounded emit: the recorded event + the honest classification.
| Field | Type | Default |
|---|---|---|
recorded | bool | required |
event | SelfObservationEvent \| None | None |
reason | str | '' |
risk | str | 'low' |
baseline_risk | str | 'low' |
requires_review | bool | False |
degraded | bool | False |
llm_used | bool | False |
anomalies | tuple[str, ...] | () |
agentic_evidence | dict \| None | None |
SelfObservationPatternRuntime¶
Stateless executable mechanisms for self_observation's applied patterns.
Methods:
tighten(risk: Any, baseline_risk: Any = '', requires_review: bool = False, degraded: bool = False, failed_tests: bool = False, irreversible_change: bool = False, human_intervention: bool = False, high_cost: bool = False) -> SelfObservationPatternReview¶
One-way risk-classification guard (hook-based-safety-guard-rails).
SelfObservationSkillCatalog¶
Maps each applied pattern slug to an event-risk classification skill record.
Methods:
list_skills() -> list[SelfObservationSkill]¶
executable_skills() -> list[SelfObservationSkill]¶
get(slug: str) -> SelfObservationSkill | None¶
Functions¶
deterministic_event_risk(event_or_payload: Any) -> RiskLevelDecision¶
Demoted-real STRUCTURAL rule classifier (the zero-LLM baseline).
apply_risk_floor(baseline: Any, candidate: Any) -> RiskLevelDecision¶
One-way risk-level safety clamp (stricter-of, ESCALATE only).
self_observation_pattern_guard(decision: RiskLevelDecision, signals: EventSignals | None = None) -> tuple[RiskLevelDecision, Any]¶
Apply the always-on one-way (escalate-only) structural guard to a decision.
grounded_classify_event(event_or_payload: Any, planner: SelfObservationPlanner | None = None) -> RiskLevelDecision¶
The ONE shared grounded chokepoint for EVERY self_observation risk decision.
emit_observation(event_type: str, payload: Mapping[str, Any], trace_root: str | Path = '.g6/traces', db_path: str | Path = 'self_observation_events.db') -> SelfObservationEmitResult¶
grounded_emit_observation(event_type: str, payload: Mapping[str, Any], planner: Any = _DEFAULT_PLANNER, trace_root: str | Path = '.g6/traces', db_path: str | Path = 'self_observation_events.db') -> 'GroundedEmitResult'¶
Module-level grounded emit: build a bus and grounded-record one event.
applied_agentic_patterns() -> list[dict[str, Any]]¶
Return compact metadata for self_observation-applied vendored patterns.
get_skill_catalog() -> SelfObservationSkillCatalog¶
MCP Tools¶
| Operation | Source |
|---|---|
ops | self_observation_mcp |
help | self_observation_mcp |
classify_event | self_observation_mcp |
emit | self_observation_mcp |
read_events | self_observation_mcp |
list_patterns | self_observation_mcp |