Skip to content

Diagnostic Collector

Diagnostic Collector — mvp.diagnostic_collector

Cluster: Uncategorised | Type: component | MCP Tools: None

Overview

Captures sanitized error diagnostics from failing components without exposing source code or user data. The DiagnosticCollectorBlock collects stack traces, resource snapshots (CPU, memory, disk), error messages, Python/platform metadata, installed component metadata, and active bypass mappings, then produces structured DiagnosticReport objects.

Supports exporting reports to JSON files, posting sanitized reports to a configured remote diagnostic sink with batching, retry queueing, exponential backoff, bearer/header/query authentication, and per-process rate limiting. Falls back to a local JSONL aggregation file when no sink is configured, with rotation controlled by G6_DIAGNOSTIC_LOCAL_MAX_BYTES.

Public operations: collect, export_file, export_sink, flush_sink_queue, auto_report, list_reports, status, ops/help, and capabilities.

Component MCP tools: diagnostics_collect, diagnostics_auto_report, diagnostics_export, diagnostics_list, diagnostics_status, diagnostics_ops, diagnostics_capabilities, and diagnostics_info.

Degradation envelope fields: degraded, degradation_reason, completion_state, warning_card, evidence, request_id, task_id, and run_id. Reliability states use the exact values verified, qualified-draft, and blocked-escalated.

The capabilities operation reports source health, optional dependencies, sink configuration, queue status, limits, rotation status, unavailable features, and the privacy-boundary limitation without sending HTTP requests.

When to use:

  • Collecting structured error diagnostics from production components
  • Sending sanitized crash reports to a central diagnostic endpoint
  • Building a local aggregate of component health data for trend analysis
  • Checking diagnostic source, sink, queue, and rotation capabilities before collection

Example:

from mvp.diagnostic_collector import DiagnosticCollectorBlock, DiagnosticInput

block = DiagnosticCollectorBlock()
result = block.infer(DiagnosticInput(
    op="collect",
    component_name="goal_engine",
    error_message="Decomposition depth exceeded",
    traceback_str="...",
))
# result.value -> DiagnosticOutput with report, sink_status, completion_state,
# warning_card, evidence, and degradation fields.

Caveats and known limitations:

  • Rate limiting is in-memory per sink URL; counters reset on process restart.
  • Queue retry uses fixed exponential backoff (max 300s) and may be slow for high-volume scenarios.
  • Relies on optional psutil for resource snapshots; degrades to empty metrics with psutil_missing when unavailable.
  • Local JSONL aggregate files rotate at the configured byte cap; rotation failure keeps appending rather than dropping diagnostics.
  • Sanitizer is pattern-based and not a formal data classification boundary; it redacts env-var patterns and common secret formats but cannot guarantee all sensitive data is caught.

Works well with: patch_client, patch_server, immune_system, observability

Public API

DiagnosticCollectorBlock(AIBlock[DiagnosticInput, DiagnosticOutput, None])

Captures sanitized error diagnostics without exposing source code or user data.

Field Type Default
name str 'diagnostic_collector'

Methods:

infer(data: DiagnosticInput) -> Result[DiagnosticOutput]

StackFrame(BaseModel)

Single frame from a sanitized stack trace.

Field Type Default
file_path str ''
line_number int 0
function_name str ''

ResourceSnapshot(BaseModel)

Point-in-time resource usage snapshot.

Field Type Default
memory_mb float 0.0
cpu_percent float 0.0
disk_free_mb float 0.0

DiagnosticReport(BaseModel)

Sanitized diagnostic report — no source code or user data.

Field Type Default
report_id str required
g6_version str '0.0.0'
timestamp str ''
component_name str ''
operation str ''
error_tag str ''
error_message str ''
stack_frames list[StackFrame] Field(default_factory=list)
input_schema dict Field(default_factory=dict)
resource_usage ResourceSnapshot Field(default_factory=ResourceSnapshot)
safety_report dict \| None None
python_version str ''
os_info str ''
platform_info str ''
installed_components list[str] Field(default_factory=list)
bypass_components dict[str, str] Field(default_factory=dict)
recent_logs list[dict] Field(default_factory=list)

DiagnosticInput(BaseModel)

Input for DiagnosticCollectorBlock.

Field Type Default
op Literal['ops', 'help', 'collect', 'export_file', 'export_sink', 'flush_sink_queue', 'auto_report', 'list_reports', 'status', 'capabilities'] required
component_name str ''
operation str ''
error_message str ''
report_id str ''
error_tag str ''
traceback_str str ''
input_model str \| None None
export_path str ''
sink_url str ''
sink_api_key str ''
sink_auth_scheme Literal['bearer', 'header', 'query', 'none'] 'bearer'
sink_auth_header str 'X-G6-Diagnostic-Key'
sink_auth_query_param str 'api_key'
local_aggregation_path str ''
allow_local_aggregation bool True
sink_queue_path str ''
sink_batch_size int 10
sink_max_retries int 5
sink_rate_limit_per_minute int 60
request_id str ''
task_id str ''
run_id str ''

DiagnosticOutput(BaseModel)

Output from DiagnosticCollectorBlock.

Field Type Default
report DiagnosticReport \| None None
reports list[dict] Field(default_factory=list)
exported_path str ''
message str ''
auto_report_summary dict Field(default_factory=dict)
available_ops list[str] Field(default_factory=list)
sink_status str ''
warning str ''
suggested_actions list[str] Field(default_factory=list)
aggregation_path str ''
queue_path str ''
queue_depth int 0
sent_count int 0
degraded bool False
degradation_reason str \| None None
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
warning_card dict[str, Any] Field(default_factory=dict)
evidence dict[str, Any] Field(default_factory=dict)
request_id str ''
task_id str ''
run_id str ''