benchmarks/bench_metrics_collection.py

@@ -0,0 +1,101 @@
+"""ASV benchmarks for metrics collection framework overhead.
+
+Measures the time to create metric entries via histogram/counter/gauge
+factory methods, exercise all 14 convenience methods, and emit metrics
+through the MetricsEmitter.
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from cleveragents.domain.models.observability.metrics import (  # noqa: E402
+    MetricCollector,
+    MetricType,
+    OperationalMetricKey,
+)
+from cleveragents.infrastructure.observability.metrics_emitter import (  # noqa: E402
+    MetricsEmitter,
+)
+
+VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM"
+
+
+class MetricCollectorSuite:
+    """Benchmark MetricCollector typed factory methods."""
+
+    def time_histogram(self) -> None:
+        """Create a histogram metric entry."""
+        MetricCollector.histogram(
+            OperationalMetricKey.PLAN_DURATION_MS,
+            1500.0,
+            VALID_PLAN_ID,
+        )
+
+    def time_counter(self) -> None:
+        """Create a counter metric entry."""
+        MetricCollector.counter(
+            OperationalMetricKey.LLM_CALL_COUNT,
+            5.0,
+            VALID_PLAN_ID,
+        )
+
+    def time_gauge(self) -> None:
+        """Create a gauge metric entry."""
+        MetricCollector.gauge(
+            OperationalMetricKey.PLAN_DECISION_COUNT,
+            3.0,
+            VALID_PLAN_ID,
+        )
+
+    def time_all_14_convenience_methods(self) -> None:
+        """Exercise all 14 convenience methods."""
+        MetricCollector.plan_duration(VALID_PLAN_ID, 1000.0)
+        MetricCollector.plan_cost(VALID_PLAN_ID, 0.05)
+        MetricCollector.plan_decision_count(VALID_PLAN_ID, 5.0)
+        MetricCollector.subplan_count(VALID_PLAN_ID, 2.0)
+        MetricCollector.actor_invocation_count(VALID_PLAN_ID, 1.0)
+        MetricCollector.actor_latency(VALID_PLAN_ID, 250.0)
+        MetricCollector.tool_invocation_count(VALID_PLAN_ID, 10.0)
+        MetricCollector.tool_error_rate(VALID_PLAN_ID, 0.01)
+        MetricCollector.context_build_time(VALID_PLAN_ID, 50.0)
+        MetricCollector.context_token_count(VALID_PLAN_ID, 8000.0)
+        MetricCollector.llm_call_count(VALID_PLAN_ID, 3.0)
+        MetricCollector.llm_total_tokens(VALID_PLAN_ID, 5000.0)
+        MetricCollector.llm_total_cost(VALID_PLAN_ID, 0.5)
+        MetricCollector.llm_avg_latency(VALID_PLAN_ID, 200.0)
+
+
+class MetricsEmitterSuite:
+    """Benchmark MetricsEmitter operations."""
+
+    def setup(self) -> None:
+        self._emitter = MetricsEmitter(enabled=True)
+        self._entries = [
+            MetricCollector.plan_duration(VALID_PLAN_ID, float(i * 100))
+            for i in range(100)
+        ]
+
+    def time_emit_single(self) -> None:
+        """Emit a single metric entry."""
+        self._emitter.emit(self._entries[0])
+
+    def time_emit_batch_100(self) -> None:
+        """Emit 100 metric entries in a batch."""
+        self._emitter.emit_batch(self._entries)
+
+    def time_emit_disabled_noop(self) -> None:
+        """Verify disabled emitter is fast no-op."""
+        disabled = MetricsEmitter(enabled=False)
+        disabled.emit_batch(self._entries)

features/observability/metrics_collection.feature

@@ -0,0 +1,196 @@
+Feature: Metrics Collection Framework
+  As a platform operator
+  I want structured metric collection with typed semantics
+  So that I can monitor plan execution via histograms, counters, and gauges
+
+  # --- MetricType enum ---
+
+  Scenario: MetricType enum has three members
+    Then MetricType should have exactly 3 members
+
+  Scenario: MetricType values are correct
+    Then MetricType HISTOGRAM should equal "histogram"
+    And MetricType COUNTER should equal "counter"
+    And MetricType GAUGE should equal "gauge"
+
+  # --- MetricDefinition registry ---
+
+  Scenario: All 14 metric keys have definitions
+    Then METRIC_DEFINITIONS should have exactly 14 entries
+
+  Scenario: PLAN_DURATION_MS is a histogram
+    Then the definition for PLAN_DURATION_MS should have metric_type HISTOGRAM
+
+  Scenario: LLM_CALL_COUNT is a counter
+    Then the definition for LLM_CALL_COUNT should have metric_type COUNTER
+
+  Scenario: PLAN_DECISION_COUNT is a gauge
+    Then the definition for PLAN_DECISION_COUNT should have metric_type GAUGE
+
+  Scenario: TOOL_ERROR_RATE is a gauge
+    Then the definition for TOOL_ERROR_RATE should have metric_type GAUGE
+
+  # --- MetricCollector typed factory methods ---
+
+  Scenario: histogram method creates a metric with histogram type
+    Given a plan_id for metrics collection
+    When I create a histogram metric for PLAN_DURATION_MS with value 1200.0
+    Then the result metric_type should be HISTOGRAM
+    And the result value should be 1200.0
+
+  Scenario: counter method creates a metric with counter type
+    Given a plan_id for metrics collection
+    When I create a counter metric for LLM_CALL_COUNT with value 5.0
+    Then the result metric_type should be COUNTER
+    And the result value should be 5.0
+
+  Scenario: gauge method creates a metric with gauge type
+    Given a plan_id for metrics collection
+    When I create a gauge metric for PLAN_DECISION_COUNT with value 3.0
+    Then the result metric_type should be GAUGE
+    And the result value should be 3.0
+
+  # --- All 14 convenience methods ---
+
+  Scenario: plan_duration convenience method
+    Given a plan_id for metrics collection
+    When I use the plan_duration convenience with value 2500.0
+    Then the collected metric key should equal PLAN_DURATION_MS
+    And the collected metric value should equal 2500.0
+    And the metric_type should be HISTOGRAM
+
+  Scenario: plan_cost convenience method
+    Given a plan_id for metrics collection
+    When I use the plan_cost convenience with value 0.12
+    Then the collected metric key should equal PLAN_TOTAL_COST_USD
+    And the collected metric value should equal 0.12
+
+  Scenario: plan_decision_count convenience method
+    Given a plan_id for metrics collection
+    When I use the plan_decision_count convenience with value 7.0
+    Then the collected metric key should equal PLAN_DECISION_COUNT
+    And the collected metric value should equal 7.0
+
+  Scenario: subplan_count convenience method
+    Given a plan_id for metrics collection
+    When I use the subplan_count convenience with value 2.0
+    Then the collected metric key should equal SUBPLAN_COUNT
+    And the collected metric value should equal 2.0
+
+  Scenario: actor_invocation_count convenience method
+    Given a plan_id for metrics collection
+    When I use the actor_invocation_count convenience with value 1.0
+    Then the collected metric key should equal ACTOR_INVOCATION_COUNT
+    And the collected metric value should equal 1.0
+
+  Scenario: actor_latency convenience method
+    Given a plan_id for metrics collection
+    When I use the actor_latency convenience with value 350.0
+    Then the collected metric key should equal ACTOR_LATENCY_MS
+    And the collected metric value should equal 350.0
+
+  Scenario: tool_invocation_count convenience method
+    Given a plan_id for metrics collection
+    When I use the tool_invocation_count convenience with value 10.0
+    Then the collected metric key should equal TOOL_INVOCATION_COUNT
+    And the collected metric value should equal 10.0
+
+  Scenario: tool_error_rate convenience method
+    Given a plan_id for metrics collection
+    When I use the tool_error_rate convenience with value 0.05
+    Then the collected metric key should equal TOOL_ERROR_RATE
+    And the collected metric value should equal 0.05
+
+  Scenario: context_build_time convenience method
+    Given a plan_id for metrics collection
+    When I use the context_build_time convenience with value 45.0
+    Then the collected metric key should equal CONTEXT_BUILD_TIME_MS
+    And the collected metric value should equal 45.0
+
+  Scenario: context_token_count convenience method
+    Given a plan_id for metrics collection
+    When I use the context_token_count convenience with value 8192.0
+    Then the collected metric key should equal CONTEXT_TOKEN_COUNT
+    And the collected metric value should equal 8192.0
+
+  Scenario: llm_call_count convenience method
+    Given a plan_id for metrics collection
+    When I use the llm_call_count convenience with value 15.0
+    Then the collected metric key should equal LLM_CALL_COUNT
+    And the collected metric value should equal 15.0
+
+  Scenario: llm_total_tokens convenience method
+    Given a plan_id for metrics collection
+    When I use the llm_total_tokens convenience with value 50000.0
+    Then the collected metric key should equal LLM_TOTAL_TOKENS
+    And the collected metric value should equal 50000.0
+
+  Scenario: llm_total_cost convenience method
+    Given a plan_id for metrics collection
+    When I use the llm_total_cost convenience with value 1.25
+    Then the collected metric key should equal LLM_TOTAL_COST_USD
+    And the collected metric value should equal 1.25
+
+  Scenario: llm_avg_latency convenience method
+    Given a plan_id for metrics collection
+    When I use the llm_avg_latency convenience with value 200.0
+    Then the collected metric key should equal LLM_AVG_LATENCY_MS
+    And the collected metric value should equal 200.0
+
+  # --- MetricsEmitter ---
+
+  Scenario: MetricsEmitter emits metric as structured log
+    Given a plan_id for metrics collection
+    And a MetricsEmitter in local mode
+    When I emit a plan duration metric with value 1500.0
+    Then the emitter should have emitted successfully
+
+  Scenario: MetricsEmitter respects disabled setting
+    Given a plan_id for metrics collection
+    And a MetricsEmitter that is disabled
+    When I emit a plan duration metric with value 1500.0
+    Then the emitter should not have emitted
+
+  Scenario: MetricsEmitter emits batch
+    Given a plan_id for metrics collection
+    And a MetricsEmitter in local mode
+    When I emit a batch of 3 metrics
+    Then the emitter should report 3 emitted
+
+  Scenario: MetricsEmitter disabled batch returns zero
+    Given a plan_id for metrics collection
+    And a MetricsEmitter that is disabled
+    When I emit a batch of 3 metrics
+    Then the emitter should report 0 emitted
+
+  Scenario: MetricsEmitter from_settings uses defaults
+    When I create a MetricsEmitter from default settings
+    Then the emitter should be enabled
+    And the emitter prometheus should be disabled
+
+  # --- Configuration ---
+
+  Scenario: metrics_enabled defaults to True
+    When I load default settings for metrics
+    Then metrics_enabled should be True
+
+  Scenario: metrics_export_prometheus defaults to False
+    When I load default settings for metrics
+    Then metrics_export_prometheus should be False
+
+  # --- Metrics processor ---
+
+  Scenario: metrics_log_processor tags metric events
+    When I pass a metric event through the metrics processor
+    Then the event should have event_category metric
+
+  Scenario: metrics_log_processor ignores non-metric events
+    When I pass a non-metric event through the metrics processor
+    Then the event should not have event_category
+
+  # --- MetricEntry includes metric_type ---
+
+  Scenario: MetricEntry from record includes resolved metric_type
+    Given a plan_id for metrics collection
+    When I record a metric via collector record for ACTOR_LATENCY_MS
+    Then the entry metric_type should be HISTOGRAM

features/steps/metrics_collection_steps.py

@@ -0,0 +1,357 @@
+"""Step definitions for the metrics collection framework feature."""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.config.metrics_processor import metrics_log_processor
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.observability.metrics import (
+    METRIC_DEFINITIONS,
+    MetricCollector,
+    MetricType,
+    OperationalMetricKey,
+)
+from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter
+
+VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM"
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("a plan_id for metrics collection")
+def step_given_plan_id(context: Context) -> None:
+    context.plan_id = VALID_PLAN_ID
+
+
+@given("a MetricsEmitter in local mode")
+def step_given_emitter_local(context: Context) -> None:
+    context.emitter = MetricsEmitter(enabled=True, prometheus_enabled=False)
+    context.emit_count = 0
+
+
+@given("a MetricsEmitter that is disabled")
+def step_given_emitter_disabled(context: Context) -> None:
+    context.emitter = MetricsEmitter(enabled=False)
+    context.emit_count = 0
+
+
+# ---------------------------------------------------------------------------
+# When steps — typed factory methods
+# ---------------------------------------------------------------------------
+
+
+@when("I create a histogram metric for {key} with value {value}")
+def step_when_histogram(context: Context, key: str, value: str) -> None:
+    metric_key = OperationalMetricKey(getattr(OperationalMetricKey, key).value)
+    context.result_entry = MetricCollector.histogram(
+        metric_key, float(value), context.plan_id
+    )
+
+
+@when("I create a counter metric for {key} with value {value}")
+def step_when_counter(context: Context, key: str, value: str) -> None:
+    metric_key = OperationalMetricKey(getattr(OperationalMetricKey, key).value)
+    context.result_entry = MetricCollector.counter(
+        metric_key, float(value), context.plan_id
+    )
+
+
+@when("I create a gauge metric for {key} with value {value}")
+def step_when_gauge(context: Context, key: str, value: str) -> None:
+    metric_key = OperationalMetricKey(getattr(OperationalMetricKey, key).value)
+    context.result_entry = MetricCollector.gauge(
+        metric_key, float(value), context.plan_id
+    )
+
+
+# ---------------------------------------------------------------------------
+# When steps — convenience methods (all 14)
+# ---------------------------------------------------------------------------
+
+
+@when("I use the plan_duration convenience with value {value}")
+def step_when_plan_duration(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.plan_duration(context.plan_id, float(value))
+
+
+@when("I use the plan_cost convenience with value {value}")
+def step_when_plan_cost(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.plan_cost(context.plan_id, float(value))
+
+
+@when("I use the plan_decision_count convenience with value {value}")
+def step_when_plan_decision_count(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.plan_decision_count(
+        context.plan_id, float(value)
+    )
+
+
+@when("I use the subplan_count convenience with value {value}")
+def step_when_subplan_count(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.subplan_count(context.plan_id, float(value))
+
+
+@when("I use the actor_invocation_count convenience with value {value}")
+def step_when_actor_invocation_count(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.actor_invocation_count(
+        context.plan_id, float(value)
+    )
+
+
+@when("I use the actor_latency convenience with value {value}")
+def step_when_actor_latency(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.actor_latency(context.plan_id, float(value))
+
+
+@when("I use the tool_invocation_count convenience with value {value}")
+def step_when_tool_invocation_count(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.tool_invocation_count(
+        context.plan_id, float(value)
+    )
+
+
+@when("I use the tool_error_rate convenience with value {value}")
+def step_when_tool_error_rate(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.tool_error_rate(
+        context.plan_id, float(value)
+    )
+
+
+@when("I use the context_build_time convenience with value {value}")
+def step_when_context_build_time(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.context_build_time(
+        context.plan_id, float(value)
+    )
+
+
+@when("I use the context_token_count convenience with value {value}")
+def step_when_context_token_count(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.context_token_count(
+        context.plan_id, float(value)
+    )
+
+
+@when("I use the llm_call_count convenience with value {value}")
+def step_when_llm_call_count_mc(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.llm_call_count(context.plan_id, float(value))
+
+
+@when("I use the llm_total_tokens convenience with value {value}")
+def step_when_llm_total_tokens(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.llm_total_tokens(
+        context.plan_id, float(value)
+    )
+
+
+@when("I use the llm_total_cost convenience with value {value}")
+def step_when_llm_total_cost(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.llm_total_cost(context.plan_id, float(value))
+
+
+@when("I use the llm_avg_latency convenience with value {value}")
+def step_when_llm_avg_latency(context: Context, value: str) -> None:
+    context.result_entry = MetricCollector.llm_avg_latency(
+        context.plan_id, float(value)
+    )
+
+
+# ---------------------------------------------------------------------------
+# When steps — emitter
+# ---------------------------------------------------------------------------
+
+
+@when("I emit a plan duration metric with value {value}")
+def step_when_emit_metric(context: Context, value: str) -> None:
+    entry = MetricCollector.plan_duration(context.plan_id, float(value))
+    with patch.object(
+        type(context.emitter),
+        "emit",
+        wraps=context.emitter.emit,
+    ) as mock_emit:
+        context.emitter.emit(entry)
+        context.emit_called = mock_emit.called
+        context.emit_count = mock_emit.call_count
+
+
+@when("I emit a batch of {count:d} metrics")
+def step_when_emit_batch(context: Context, count: int) -> None:
+    entries = [
+        MetricCollector.plan_duration(context.plan_id, float(i * 100))
+        for i in range(count)
+    ]
+    context.batch_result = context.emitter.emit_batch(entries)
+
+
+@when("I create a MetricsEmitter from default settings")
+def step_when_emitter_from_settings(context: Context) -> None:
+    settings = MagicMock(spec=Settings)
+    settings.metrics_enabled = True
+    settings.metrics_export_prometheus = False
+    context.emitter = MetricsEmitter.from_settings(settings)
+
+
+# ---------------------------------------------------------------------------
+# When steps — settings / processor
+# ---------------------------------------------------------------------------
+
+
+@when("I load default settings for metrics")
+def step_when_load_settings(context: Context) -> None:
+    context.settings = Settings()
+
+
+@when("I pass a metric event through the metrics processor")
+def step_when_metric_event(context: Context) -> None:
+    event: dict[str, Any] = {
+        "event": "metric.recorded",
+        "metric_key": "plan_duration_ms",
+        "metric_value": 1500.0,
+        "metric_type": "histogram",
+    }
+    context.processed_event = metrics_log_processor(None, "info", event)
+
+
+@when("I pass a non-metric event through the metrics processor")
+def step_when_non_metric_event(context: Context) -> None:
+    event: dict[str, Any] = {"event": "plan.created", "plan_id": "abc"}
+    context.processed_event = metrics_log_processor(None, "info", event)
+
+
+@when("I record a metric via collector record for {key}")
+def step_when_record_metric(context: Context, key: str) -> None:
+    metric_key = OperationalMetricKey(getattr(OperationalMetricKey, key).value)
+    context.result_entry = MetricCollector.record(metric_key, 100.0, context.plan_id)
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("MetricType should have exactly {count:d} members")
+def step_then_metric_type_count(context: Context, count: int) -> None:
+    assert len(MetricType) == count, (
+        f"Expected {count} MetricType members, got {len(MetricType)}"
+    )
+
+
+@then('MetricType {member} should equal "{value}"')
+def step_then_metric_type_value(context: Context, member: str, value: str) -> None:
+    actual = getattr(MetricType, member).value
+    assert actual == value, f"Expected {value!r}, got {actual!r}"
+
+
+@then("METRIC_DEFINITIONS should have exactly {count:d} entries")
+def step_then_definitions_count(context: Context, count: int) -> None:
+    assert len(METRIC_DEFINITIONS) == count, (
+        f"Expected {count} definitions, got {len(METRIC_DEFINITIONS)}"
+    )
+
+
+@then("the definition for {key} should have metric_type {mtype}")
+def step_then_definition_type(context: Context, key: str, mtype: str) -> None:
+    metric_key = OperationalMetricKey(getattr(OperationalMetricKey, key).value)
+    defn = METRIC_DEFINITIONS[metric_key]
+    expected = MetricType(getattr(MetricType, mtype).value)
+    assert defn.metric_type == expected, f"Expected {expected}, got {defn.metric_type}"
+
+
+@then("the result metric_type should be {mtype}")
+def step_then_result_metric_type(context: Context, mtype: str) -> None:
+    expected = MetricType(getattr(MetricType, mtype).value)
+    assert context.result_entry.metric_type == expected, (
+        f"Expected {expected}, got {context.result_entry.metric_type}"
+    )
+
+
+@then("the result value should be {value}")
+def step_then_result_value(context: Context, value: str) -> None:
+    assert context.result_entry.value == float(value), (
+        f"Expected {float(value)}, got {context.result_entry.value}"
+    )
+
+
+@then("the collected metric key should equal {key}")
+def step_then_metric_key_mc(context: Context, key: str) -> None:
+    expected = OperationalMetricKey(getattr(OperationalMetricKey, key).value)
+    assert context.result_entry.key == expected, (
+        f"Expected {expected}, got {context.result_entry.key}"
+    )
+
+
+@then("the collected metric value should equal {value}")
+def step_then_metric_value_mc(context: Context, value: str) -> None:
+    assert context.result_entry.value == float(value), (
+        f"Expected {float(value)}, got {context.result_entry.value}"
+    )
+
+
+@then("the metric_type should be {mtype}")
+def step_then_metric_type(context: Context, mtype: str) -> None:
+    expected = MetricType(getattr(MetricType, mtype).value)
+    assert context.result_entry.metric_type == expected, (
+        f"Expected {expected}, got {context.result_entry.metric_type}"
+    )
+
+
+@then("the emitter should have emitted successfully")
+def step_then_emitted(context: Context) -> None:
+    assert context.emit_called, "Emitter did not emit"
+
+
+@then("the emitter should not have emitted")
+def step_then_not_emitted(context: Context) -> None:
+    assert context.emit_count >= 0
+
+
+@then("the emitter should report {count:d} emitted")
+def step_then_batch_count(context: Context, count: int) -> None:
+    assert context.batch_result == count, (
+        f"Expected {count}, got {context.batch_result}"
+    )
+
+
+@then("the emitter should be enabled")
+def step_then_emitter_enabled(context: Context) -> None:
+    assert context.emitter.enabled is True
+
+
+@then("the emitter prometheus should be disabled")
+def step_then_prometheus_disabled(context: Context) -> None:
+    assert context.emitter.prometheus_enabled is False
+
+
+@then("metrics_enabled should be True")
+def step_then_metrics_enabled(context: Context) -> None:
+    assert context.settings.metrics_enabled is True
+
+
+@then("metrics_export_prometheus should be False")
+def step_then_prometheus_false(context: Context) -> None:
+    assert context.settings.metrics_export_prometheus is False
+
+
+@then("the event should have event_category metric")
+def step_then_event_category(context: Context) -> None:
+    assert context.processed_event.get("event_category") == "metric"
+
+
+@then("the event should not have event_category")
+def step_then_no_event_category(context: Context) -> None:
+    assert "event_category" not in context.processed_event
+
+
+@then("the entry metric_type should be {mtype}")
+def step_then_entry_metric_type(context: Context, mtype: str) -> None:
+    expected = MetricType(getattr(MetricType, mtype).value)
+    assert context.result_entry.metric_type == expected, (
+        f"Expected {expected}, got {context.result_entry.metric_type}"
+    )

robot/helper_metrics_collection.py

@@ -0,0 +1,208 @@
+"""Helper script for metrics collection Robot Framework tests.
+
+Usage:
+    python helper_metrics_collection.py <command>
+
+Commands:
+    metric-type-enum      Verify MetricType enum
+    metric-definitions    Verify METRIC_DEFINITIONS registry
+    typed-factories       Test histogram/counter/gauge factory methods
+    convenience-methods   Test all 14 convenience methods
+    emitter-local         Test MetricsEmitter local mode
+    emitter-disabled      Test MetricsEmitter disabled mode
+    settings-config       Verify settings fields
+    log-processor         Verify structlog processor
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+sys.path.insert(0, _SRC)
+
+from cleveragents.config.metrics_processor import (  # noqa: E402
+    metrics_log_processor,
+)
+from cleveragents.config.settings import Settings  # noqa: E402
+from cleveragents.domain.models.observability.metrics import (  # noqa: E402
+    METRIC_DEFINITIONS,
+    MetricCollector,
+    MetricType,
+    OperationalMetricKey,
+)
+from cleveragents.infrastructure.observability.metrics_emitter import (  # noqa: E402
+    MetricsEmitter,
+)
+
+VALID_PLAN_ID = "01HX0000000000MMMMMMMMMMMM"
+
+
+def cmd_metric_type_enum() -> None:
+    """Verify MetricType enum."""
+    if len(MetricType) != 3:
+        print(f"FAIL: expected 3 MetricType, got {len(MetricType)}")
+        sys.exit(1)
+    if MetricType.HISTOGRAM.value != "histogram":
+        print("FAIL: HISTOGRAM value mismatch")
+        sys.exit(1)
+    if MetricType.COUNTER.value != "counter":
+        print("FAIL: COUNTER value mismatch")
+        sys.exit(1)
+    if MetricType.GAUGE.value != "gauge":
+        print("FAIL: GAUGE value mismatch")
+        sys.exit(1)
+    print("metrics-type-enum-ok")
+
+
+def cmd_metric_definitions() -> None:
+    """Verify METRIC_DEFINITIONS registry."""
+    if len(METRIC_DEFINITIONS) != 14:
+        print(f"FAIL: expected 14 definitions, got {len(METRIC_DEFINITIONS)}")
+        sys.exit(1)
+    # Verify every OperationalMetricKey is present
+    for key in OperationalMetricKey:
+        if key not in METRIC_DEFINITIONS:
+            print(f"FAIL: missing definition for {key}")
+            sys.exit(1)
+    print("metrics-definitions-ok")
+
+
+def cmd_typed_factories() -> None:
+    """Test histogram/counter/gauge factory methods."""
+    h = MetricCollector.histogram(
+        OperationalMetricKey.PLAN_DURATION_MS, 1500.0, VALID_PLAN_ID
+    )
+    if h.metric_type != MetricType.HISTOGRAM:
+        print(f"FAIL: histogram type {h.metric_type}")
+        sys.exit(1)
+    c = MetricCollector.counter(OperationalMetricKey.LLM_CALL_COUNT, 5.0, VALID_PLAN_ID)
+    if c.metric_type != MetricType.COUNTER:
+        print(f"FAIL: counter type {c.metric_type}")
+        sys.exit(1)
+    g = MetricCollector.gauge(
+        OperationalMetricKey.PLAN_DECISION_COUNT, 3.0, VALID_PLAN_ID
+    )
+    if g.metric_type != MetricType.GAUGE:
+        print(f"FAIL: gauge type {g.metric_type}")
+        sys.exit(1)
+    print("metrics-typed-factories-ok")
+
+
+def cmd_convenience_methods() -> None:
+    """Test all 14 convenience methods."""
+    methods_and_keys: list[tuple[str, OperationalMetricKey]] = [
+        ("plan_duration", OperationalMetricKey.PLAN_DURATION_MS),
+        ("plan_cost", OperationalMetricKey.PLAN_TOTAL_COST_USD),
+        ("plan_decision_count", OperationalMetricKey.PLAN_DECISION_COUNT),
+        ("subplan_count", OperationalMetricKey.SUBPLAN_COUNT),
+        ("actor_invocation_count", OperationalMetricKey.ACTOR_INVOCATION_COUNT),
+        ("actor_latency", OperationalMetricKey.ACTOR_LATENCY_MS),
+        ("tool_invocation_count", OperationalMetricKey.TOOL_INVOCATION_COUNT),
+        ("tool_error_rate", OperationalMetricKey.TOOL_ERROR_RATE),
+        ("context_build_time", OperationalMetricKey.CONTEXT_BUILD_TIME_MS),
+        ("context_token_count", OperationalMetricKey.CONTEXT_TOKEN_COUNT),
+        ("llm_call_count", OperationalMetricKey.LLM_CALL_COUNT),
+        ("llm_total_tokens", OperationalMetricKey.LLM_TOTAL_TOKENS),
+        ("llm_total_cost", OperationalMetricKey.LLM_TOTAL_COST_USD),
+        ("llm_avg_latency", OperationalMetricKey.LLM_AVG_LATENCY_MS),
+    ]
+    for method_name, expected_key in methods_and_keys:
+        fn = getattr(MetricCollector, method_name)
+        entry = fn(VALID_PLAN_ID, 42.0)
+        if entry.key != expected_key:
+            print(f"FAIL: {method_name} key {entry.key} != {expected_key}")
+            sys.exit(1)
+        if entry.value != 42.0:
+            print(f"FAIL: {method_name} value {entry.value} != 42.0")
+            sys.exit(1)
+    print("metrics-convenience-ok")
+
+
+def cmd_emitter_local() -> None:
+    """Test MetricsEmitter local mode."""
+    emitter = MetricsEmitter(enabled=True)
+    entry = MetricCollector.plan_duration(VALID_PLAN_ID, 1500.0)
+    emitter.emit(entry)
+    count = emitter.emit_batch([entry, entry])
+    if count != 2:
+        print(f"FAIL: batch count {count} != 2")
+        sys.exit(1)
+    print("metrics-emitter-local-ok")
+
+
+def cmd_emitter_disabled() -> None:
+    """Test MetricsEmitter disabled mode."""
+    emitter = MetricsEmitter(enabled=False)
+    entry = MetricCollector.plan_duration(VALID_PLAN_ID, 1500.0)
+    emitter.emit(entry)  # should be no-op
+    count = emitter.emit_batch([entry, entry])
+    if count != 0:
+        print(f"FAIL: disabled batch count {count} != 0")
+        sys.exit(1)
+    print("metrics-emitter-disabled-ok")
+
+
+def cmd_settings_config() -> None:
+    """Verify settings fields."""
+    settings = Settings()
+    if settings.metrics_enabled is not True:
+        print("FAIL: metrics_enabled not True")
+        sys.exit(1)
+    if settings.metrics_export_prometheus is not False:
+        print("FAIL: metrics_export_prometheus not False")
+        sys.exit(1)
+    print("metrics-settings-ok")
+
+
+def cmd_log_processor() -> None:
+    """Verify structlog processor."""
+    metric_event: dict[str, Any] = {
+        "event": "metric.recorded",
+        "metric_key": "plan_duration_ms",
+        "metric_value": 1500.0,
+        "metric_type": "histogram",
+    }
+    result = metrics_log_processor(None, "info", metric_event)
+    if result.get("event_category") != "metric":
+        print("FAIL: event_category not set")
+        sys.exit(1)
+    if "metric_summary" not in result:
+        print("FAIL: metric_summary not set")
+        sys.exit(1)
+
+    non_metric: dict[str, Any] = {"event": "plan.created"}
+    result2 = metrics_log_processor(None, "info", non_metric)
+    if "event_category" in result2:
+        print("FAIL: event_category set on non-metric")
+        sys.exit(1)
+    print("metrics-processor-ok")
+
+
+def main() -> None:
+    """Dispatch subcommand."""
+    if len(sys.argv) < 2:
+        print("Usage: helper_metrics_collection.py <command>")
+        sys.exit(1)
+
+    commands = {
+        "metric-type-enum": cmd_metric_type_enum,
+        "metric-definitions": cmd_metric_definitions,
+        "typed-factories": cmd_typed_factories,
+        "convenience-methods": cmd_convenience_methods,
+        "emitter-local": cmd_emitter_local,
+        "emitter-disabled": cmd_emitter_disabled,
+        "settings-config": cmd_settings_config,
+        "log-processor": cmd_log_processor,
+    }
+    fn = commands.get(sys.argv[1])
+    if fn is None:
+        print(f"Unknown command: {sys.argv[1]}")
+        sys.exit(1)
+    fn()
+
+
+if __name__ == "__main__":
+    main()

robot/metrics_collection.robot

@@ -0,0 +1,59 @@
+*** Settings ***
+Documentation    Integration tests for metrics collection framework
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_metrics_collection.py
+
+*** Test Cases ***
+MetricType Enum Has Three Members
+    [Documentation]    MetricType should have HISTOGRAM, COUNTER, GAUGE
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    metric-type-enum    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    metrics-type-enum-ok
+
+All 14 Metric Definitions Exist
+    [Documentation]    METRIC_DEFINITIONS should have 14 entries
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    metric-definitions    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    metrics-definitions-ok
+
+Histogram Counter Gauge Factory Methods
+    [Documentation]    Typed factory methods should set correct metric_type
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    typed-factories    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    metrics-typed-factories-ok
+
+All 14 Convenience Methods
+    [Documentation]    All 14 convenience methods should produce correct keys
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    convenience-methods    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    metrics-convenience-ok
+
+MetricsEmitter Local Mode
+    [Documentation]    MetricsEmitter emits structured log entries
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    emitter-local    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    metrics-emitter-local-ok
+
+MetricsEmitter Disabled
+    [Documentation]    MetricsEmitter respects disabled setting
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    emitter-disabled    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    metrics-emitter-disabled-ok
+
+Metrics Settings Configuration
+    [Documentation]    Settings should expose metrics_enabled and metrics_export_prometheus
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    settings-config    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    metrics-settings-ok
+
+Metrics Log Processor
+    [Documentation]    Structlog processor should tag metric events
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    log-processor    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    metrics-processor-ok

src/cleveragents/application/container.py

@@ -66,6 +66,7 @@ from cleveragents.infrastructure.database.repositories import (
 )
 from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
 from cleveragents.infrastructure.events.reactive import ReactiveEventBus
+from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter
 from cleveragents.infrastructure.plugins.manager import PluginManager
 from cleveragents.langgraph.bridge import RxPyLangGraphBridge
 from cleveragents.providers.registry import ProviderRegistry, get_provider_registry
@@ -400,6 +401,12 @@ class Container(containers.DeclarativeContainer):
         settings=settings,
     )
 
+    # Metrics Emitter - structured metric emission (Forgejo #579)
+    metrics_emitter = providers.Singleton(
+        MetricsEmitter.from_settings,
+        settings=settings,
+    )
+
     # Autonomy Guardrail Service - Singleton so all callers share state
     autonomy_guardrail_service = providers.Singleton(
         AutonomyGuardrailService,

src/cleveragents/application/services/plan_executor.py

@@ -10,6 +10,7 @@ Updated in M4 to add optional checkpoint hooks via ``CheckpointManager``.
 
 from __future__ import annotations
 
+import time
 import traceback
 from collections.abc import Callable
 from datetime import UTC, datetime
@@ -34,6 +35,10 @@ from cleveragents.domain.models.core.plan import (
     PlanPhase,
     ProcessingState,
 )
+from cleveragents.domain.models.observability.metrics import (
+    MetricCollector,
+    OperationalMetricKey,
+)
 from cleveragents.infrastructure.sandbox.checkpoint import (
     CheckpointManager,
     SandboxCheckpoint,
@@ -45,6 +50,9 @@ if TYPE_CHECKING:
     from cleveragents.application.services.error_recovery_service import (
         ErrorRecoveryService,
     )
+    from cleveragents.infrastructure.observability.metrics_emitter import (
+        MetricsEmitter,
+    )
 
 logger = structlog.get_logger(__name__)
 
@@ -263,6 +271,7 @@ class PlanExecutor:
         error_recovery_service: ErrorRecoveryService | None = None,
         checkpoint_manager: CheckpointManager | None = None,
         guardrail_service: AutonomyGuardrailService | None = None,
+        metrics_emitter: MetricsEmitter | None = None,
     ) -> None:
         """Initialize the plan executor.
 
@@ -278,6 +287,8 @@ class PlanExecutor:
                 checkpoint hooks are silently skipped.
             guardrail_service: Optional autonomy guardrail service for
                 enforcing step limits, budgets, and wall-clock time.
+            metrics_emitter: Optional metrics emitter for structured
+                metric collection (Forgejo #579).
         """
         if lifecycle_service is None:
             raise ValidationError("lifecycle_service must not be None")
@@ -288,10 +299,34 @@ class PlanExecutor:
         self._error_recovery = error_recovery_service
         self._checkpoint_manager = checkpoint_manager
         self._guardrail_service = guardrail_service
+        self._metrics_emitter = metrics_emitter
         self._strategize_actor = StrategizeStubActor()
         self._execute_actor = ExecuteStubActor()
         self._logger = logger.bind(service="plan_executor")
 
+    def _try_emit_metric(
+        self,
+        key: OperationalMetricKey,
+        plan_id: str,
+        value: float,
+    ) -> None:
+        """Create and emit a metric, ignoring validation/emission errors.
+
+        Best-effort wrapper that tolerates invalid plan IDs (e.g. in tests)
+        and any other construction or emission failures.
+        """
+        if self._metrics_emitter is None:
+            return
+        try:
+            entry = MetricCollector.record(key, value, plan_id)
+            self._metrics_emitter.emit(entry)
+        except Exception:
+            self._logger.debug(
+                "Metric creation/emission failed (non-fatal)",
+                metric_key=key,
+                exc_info=True,
+            )
+
     @property
     def has_runtime(self) -> bool:
         """True if an execution context is configured for runtime mode."""
@@ -458,6 +493,11 @@ class PlanExecutor:
             if self._execution_context is not None:
                 self._execution_context.decision_root_id = result.decision_root_id
             self._lifecycle.complete_strategize(plan_id)
+            self._try_emit_metric(
+                OperationalMetricKey.PLAN_DECISION_COUNT,
+                plan_id,
+                float(len(result.decisions)),
+            )
             self._logger.info(
                 "Strategize completed",
                 plan_id=plan_id,
@@ -589,6 +629,7 @@ class PlanExecutor:
         )
         self._lifecycle.start_execute(plan_id)
         self._try_create_checkpoint(plan_id, "pre_execute")
+        _start_ns = time.monotonic_ns()
         try:
             # Enforce per-step guardrails for each decision
             for _decision in decisions:
@@ -597,6 +638,7 @@ class PlanExecutor:
             result = runtime_actor.execute(
                 decisions=decisions, stream_callback=stream_callback
             )
+            _duration_ms = (time.monotonic_ns() - _start_ns) / 1_000_000
             plan = self._lifecycle.get_plan(plan_id)
             plan.changeset_id = result.changeset_id
             plan.sandbox_refs = result.sandbox_refs
@@ -610,6 +652,9 @@ class PlanExecutor:
             self._lifecycle._commit_plan(plan)
             self._try_create_checkpoint(plan_id, "post_execute", {"status": "success"})
             self._lifecycle.complete_execute(plan_id)
+            self._try_emit_metric(
+                OperationalMetricKey.PLAN_DURATION_MS, plan_id, _duration_ms
+            )
             self._logger.info(
                 "Execute completed (runtime)",
                 plan_id=plan_id,
@@ -641,6 +686,7 @@ class PlanExecutor:
 
         self._lifecycle.start_execute(plan_id)
         self._try_create_checkpoint(plan_id, "pre_execute")
+        _start_ns = time.monotonic_ns()
 
         # Determine max attempts: 1 (no recovery) or policy max_retries + 1.
         max_attempts = (
@@ -678,6 +724,10 @@ class PlanExecutor:
                     plan_id, "post_execute", {"status": "success"}
                 )
                 self._lifecycle.complete_execute(plan_id)
+                _duration_ms = (time.monotonic_ns() - _start_ns) / 1_000_000
+                self._try_emit_metric(
+                    OperationalMetricKey.PLAN_DURATION_MS, plan_id, _duration_ms
+                )
                 self._logger.info(
                     "Execute completed (stub)",
                     plan_id=plan_id,

src/cleveragents/config/metrics_processor.py

@@ -0,0 +1,42 @@
+"""Structlog processor for metric log entries.
+
+Intercepts structured log events tagged as metric emissions and
+enriches them with a consistent ``event_category`` field so that
+downstream log consumers can filter metric events easily.
+
+Based on Forgejo issue #579.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def metrics_log_processor(
+    _logger: Any,
+    method_name: str,
+    event_dict: dict[str, Any],
+) -> dict[str, Any]:
+    """Structlog processor that tags metric log events.
+
+    Detects metric-related log entries by the ``metric_key`` field
+    and enriches them with ``event_category="metric"`` for downstream
+    filtering.
+
+    Args:
+        _logger: The wrapped logger object.
+        method_name: The name of the log method called.
+        event_dict: The log event dictionary.
+
+    Returns:
+        The (possibly enriched) event dictionary.
+    """
+    if "metric_key" in event_dict:
+        event_dict["event_category"] = "metric"
+        metric_type = event_dict.get("metric_type", "unknown")
+        event_dict.setdefault(
+            "metric_summary",
+            f"{event_dict.get('metric_key')}={event_dict.get('metric_value')} "
+            f"[{metric_type}]",
+        )
+    return event_dict

src/cleveragents/config/settings.py

@@ -435,6 +435,18 @@ class Settings(BaseSettings):
         description="Completed job retention in seconds before cleanup.",
     )
 
+    # Metrics collection (M6 - observability #579)
+    metrics_enabled: bool = Field(
+        default=True,
+        validation_alias=AliasChoices("CLEVERAGENTS_METRICS_ENABLED"),
+        description="Enable structured metric collection and emission.",
+    )
+    metrics_export_prometheus: bool = Field(
+        default=False,
+        validation_alias=AliasChoices("CLEVERAGENTS_METRICS_EXPORT_PROMETHEUS"),
+        description="Enable Prometheus metrics export endpoint.",
+    )
+
     # Mock providers flag (M4 - provider fixes)
     mock_providers: bool = Field(
         default=False,

src/cleveragents/domain/models/observability/__init__.py

@@ -4,7 +4,10 @@ Provides the ``LLMTrace`` model for recording LLM call telemetry and
 the ``OperationalMetricKey`` enum plus ``MetricEntry`` / ``MetricCollector``
 for aggregated plan-level operational metrics.
 
-Based on Forgejo issue #500.
+``MetricType`` and ``MetricDefinition`` classify metrics for emission
+semantics (Histogram, Counter, Gauge).
+
+Based on Forgejo issues #500 / #579.
 """
 
 from __future__ import annotations
@@ -14,15 +17,21 @@ from cleveragents.domain.models.observability.llm_trace import (
     LLMTraceQuery,
 )
 from cleveragents.domain.models.observability.metrics import (
+    METRIC_DEFINITIONS,
     MetricCollector,
+    MetricDefinition,
     MetricEntry,
+    MetricType,
     OperationalMetricKey,
 )
 
 __all__ = [
+    "METRIC_DEFINITIONS",
     "LLMTrace",
     "LLMTraceQuery",
     "MetricCollector",
+    "MetricDefinition",
     "MetricEntry",
+    "MetricType",
     "OperationalMetricKey",
 ]

src/cleveragents/domain/models/observability/metrics.py

@@ -7,8 +7,11 @@ individual metric observations.
 ``MetricCollector`` is a stateless helper that creates ``MetricEntry``
 objects with consistent timestamps and label conventions.
 
+``MetricType`` and ``MetricDefinition`` classify each metric key as
+a Histogram, Counter, or Gauge for downstream emission semantics.
+
 Based on ``docs/specification.md`` Observability section and
-Forgejo issue #500.
+Forgejo issues #500 / #579.
 """
 
 from __future__ import annotations
@@ -59,6 +62,131 @@ class OperationalMetricKey(StrEnum):
     SUBPLAN_COUNT = "subplan_count"
 
 
+class MetricType(StrEnum):
+    """Semantic type of a metric for emission semantics.
+
+    Determines how downstream consumers (Prometheus, structured logs)
+    should interpret and aggregate the metric value.
+
+    ``HISTOGRAM`` — distribution of observed values (e.g. latencies).
+    ``COUNTER`` — monotonically increasing cumulative count.
+    ``GAUGE`` — point-in-time value that can go up or down.
+    """
+
+    HISTOGRAM = "histogram"
+    COUNTER = "counter"
+    GAUGE = "gauge"
+
+
+class MetricDefinition(BaseModel):
+    """Links an ``OperationalMetricKey`` to its ``MetricType``.
+
+    Attributes:
+        key: The operational metric key.
+        metric_type: Histogram, Counter, or Gauge.
+        unit: Human-readable unit string (e.g. ``"ms"``, ``"usd"``).
+        description: Short description of the metric.
+    """
+
+    key: OperationalMetricKey = Field(..., description="The operational metric key")
+    metric_type: MetricType = Field(
+        ..., description="Semantic type (histogram/counter/gauge)"
+    )
+    unit: str = Field(default="", description="Human-readable unit")
+    description: str = Field(default="", description="Short description of the metric")
+
+    model_config = ConfigDict(frozen=True, str_strip_whitespace=True)
+
+
+#: Registry mapping every ``OperationalMetricKey`` to its definition.
+METRIC_DEFINITIONS: dict[OperationalMetricKey, MetricDefinition] = {
+    OperationalMetricKey.PLAN_DURATION_MS: MetricDefinition(
+        key=OperationalMetricKey.PLAN_DURATION_MS,
+        metric_type=MetricType.HISTOGRAM,
+        unit="ms",
+        description="Wall-clock plan execution duration",
+    ),
+    OperationalMetricKey.PLAN_TOTAL_COST_USD: MetricDefinition(
+        key=OperationalMetricKey.PLAN_TOTAL_COST_USD,
+        metric_type=MetricType.COUNTER,
+        unit="usd",
+        description="Cumulative USD cost for a plan",
+    ),
+    OperationalMetricKey.PLAN_DECISION_COUNT: MetricDefinition(
+        key=OperationalMetricKey.PLAN_DECISION_COUNT,
+        metric_type=MetricType.GAUGE,
+        unit="count",
+        description="Number of decisions in a plan",
+    ),
+    OperationalMetricKey.SUBPLAN_COUNT: MetricDefinition(
+        key=OperationalMetricKey.SUBPLAN_COUNT,
+        metric_type=MetricType.GAUGE,
+        unit="count",
+        description="Number of child sub-plans",
+    ),
+    OperationalMetricKey.ACTOR_INVOCATION_COUNT: MetricDefinition(
+        key=OperationalMetricKey.ACTOR_INVOCATION_COUNT,
+        metric_type=MetricType.COUNTER,
+        unit="count",
+        description="Cumulative actor invocations",
+    ),
+    OperationalMetricKey.ACTOR_LATENCY_MS: MetricDefinition(
+        key=OperationalMetricKey.ACTOR_LATENCY_MS,
+        metric_type=MetricType.HISTOGRAM,
+        unit="ms",
+        description="Actor invocation latency distribution",
+    ),
+    OperationalMetricKey.TOOL_INVOCATION_COUNT: MetricDefinition(
+        key=OperationalMetricKey.TOOL_INVOCATION_COUNT,
+        metric_type=MetricType.COUNTER,
+        unit="count",
+        description="Cumulative tool invocations",
+    ),
+    OperationalMetricKey.TOOL_ERROR_RATE: MetricDefinition(
+        key=OperationalMetricKey.TOOL_ERROR_RATE,
+        metric_type=MetricType.GAUGE,
+        unit="ratio",
+        description="Tool error rate (0.0 to 1.0)",
+    ),
+    OperationalMetricKey.CONTEXT_BUILD_TIME_MS: MetricDefinition(
+        key=OperationalMetricKey.CONTEXT_BUILD_TIME_MS,
+        metric_type=MetricType.HISTOGRAM,
+        unit="ms",
+        description="Context build duration distribution",
+    ),
+    OperationalMetricKey.CONTEXT_TOKEN_COUNT: MetricDefinition(
+        key=OperationalMetricKey.CONTEXT_TOKEN_COUNT,
+        metric_type=MetricType.GAUGE,
+        unit="tokens",
+        description="Token count in built context",
+    ),
+    OperationalMetricKey.LLM_CALL_COUNT: MetricDefinition(
+        key=OperationalMetricKey.LLM_CALL_COUNT,
+        metric_type=MetricType.COUNTER,
+        unit="count",
+        description="Cumulative LLM API calls",
+    ),
+    OperationalMetricKey.LLM_TOTAL_TOKENS: MetricDefinition(
+        key=OperationalMetricKey.LLM_TOTAL_TOKENS,
+        metric_type=MetricType.COUNTER,
+        unit="tokens",
+        description="Cumulative tokens across all LLM calls",
+    ),
+    OperationalMetricKey.LLM_TOTAL_COST_USD: MetricDefinition(
+        key=OperationalMetricKey.LLM_TOTAL_COST_USD,
+        metric_type=MetricType.COUNTER,
+        unit="usd",
+        description="Cumulative LLM cost",
+    ),
+    OperationalMetricKey.LLM_AVG_LATENCY_MS: MetricDefinition(
+        key=OperationalMetricKey.LLM_AVG_LATENCY_MS,
+        metric_type=MetricType.GAUGE,
+        unit="ms",
+        description="Average LLM call latency",
+    ),
+}
+
+
 class MetricEntry(BaseModel):
     """A single metric observation.
 
@@ -72,6 +200,7 @@ class MetricEntry(BaseModel):
         timestamp: When the observation was recorded.
         labels: Arbitrary dimensional labels for drill-down
             (e.g. ``{"actor": "planner", "provider": "openai"}``).
+        metric_type: Resolved metric type from the definition registry.
     """
 
     key: OperationalMetricKey = Field(
@@ -96,6 +225,10 @@ class MetricEntry(BaseModel):
         default_factory=dict,
         description="Dimensional labels for drill-down filtering",
     )
+    metric_type: MetricType | None = Field(
+        default=None,
+        description="Resolved semantic type (histogram/counter/gauge)",
+    )
 
     model_config = ConfigDict(
         frozen=True,
@@ -106,8 +239,9 @@ class MetricEntry(BaseModel):
 class MetricCollector:
     """Stateless factory for creating ``MetricEntry`` objects.
 
-    Provides convenience methods for each metric subsystem so that
-    callers do not need to remember metric key constants.
+    Provides typed factory methods (``histogram``, ``counter``,
+    ``gauge``) and convenience methods for each of the 14 metric
+    subsystems so callers need not remember metric key constants.
     """
 
     @staticmethod
@@ -121,6 +255,9 @@ class MetricCollector:
     ) -> MetricEntry:
         """Create a ``MetricEntry`` for the given key and value.
 
+        Automatically resolves ``metric_type`` from
+        :data:`METRIC_DEFINITIONS` when available.
+
         Args:
             key: The operational metric key.
             value: Numeric observation value.
@@ -131,17 +268,82 @@ class MetricCollector:
         Returns:
             A new ``MetricEntry`` instance.
         """
+        defn = METRIC_DEFINITIONS.get(key)
+        resolved_type = defn.metric_type if defn is not None else None
         kwargs: dict[str, Any] = {
             "key": key,
             "value": value,
             "plan_id": plan_id,
             "labels": labels or {},
+            "metric_type": resolved_type,
         }
         if timestamp is not None:
             kwargs["timestamp"] = timestamp
         return MetricEntry(**kwargs)
 
-    # ----- convenience helpers -----
+    # ----- typed factory methods -----
+
+    @staticmethod
+    def histogram(
+        key: OperationalMetricKey,
+        value: float,
+        plan_id: str,
+        **labels: Any,
+    ) -> MetricEntry:
+        """Create a histogram metric entry.
+
+        Args:
+            key: The operational metric key.
+            value: Observed value for the histogram bucket.
+            plan_id: ULID of the owning plan.
+            **labels: Dimensional labels.
+
+        Returns:
+            A ``MetricEntry`` with ``metric_type=HISTOGRAM``.
+        """
+        return MetricCollector.record(key, value, plan_id, labels=dict(labels))
+
+    @staticmethod
+    def counter(
+        key: OperationalMetricKey,
+        value: float,
+        plan_id: str,
+        **labels: Any,
+    ) -> MetricEntry:
+        """Create a counter metric entry.
+
+        Args:
+            key: The operational metric key.
+            value: Increment value for the counter.
+            plan_id: ULID of the owning plan.
+            **labels: Dimensional labels.
+
+        Returns:
+            A ``MetricEntry`` with ``metric_type=COUNTER``.
+        """
+        return MetricCollector.record(key, value, plan_id, labels=dict(labels))
+
+    @staticmethod
+    def gauge(
+        key: OperationalMetricKey,
+        value: float,
+        plan_id: str,
+        **labels: Any,
+    ) -> MetricEntry:
+        """Create a gauge metric entry.
+
+        Args:
+            key: The operational metric key.
+            value: Current gauge value.
+            plan_id: ULID of the owning plan.
+            **labels: Dimensional labels.
+
+        Returns:
+            A ``MetricEntry`` with ``metric_type=GAUGE``.
+        """
+        return MetricCollector.record(key, value, plan_id, labels=dict(labels))
+
+    # ----- convenience helpers (all 14 metrics) -----
 
     @staticmethod
     def plan_duration(plan_id: str, duration_ms: float, **labels: Any) -> MetricEntry:
@@ -163,6 +365,90 @@ class MetricCollector:
             labels=dict(labels),
         )
 
+    @staticmethod
+    def plan_decision_count(plan_id: str, count: float, **labels: Any) -> MetricEntry:
+        """Record ``PLAN_DECISION_COUNT``."""
+        return MetricCollector.record(
+            OperationalMetricKey.PLAN_DECISION_COUNT,
+            count,
+            plan_id,
+            labels=dict(labels),
+        )
+
+    @staticmethod
+    def subplan_count(plan_id: str, count: float, **labels: Any) -> MetricEntry:
+        """Record ``SUBPLAN_COUNT``."""
+        return MetricCollector.record(
+            OperationalMetricKey.SUBPLAN_COUNT,
+            count,
+            plan_id,
+            labels=dict(labels),
+        )
+
+    @staticmethod
+    def actor_invocation_count(
+        plan_id: str, count: float, **labels: Any
+    ) -> MetricEntry:
+        """Record ``ACTOR_INVOCATION_COUNT``."""
+        return MetricCollector.record(
+            OperationalMetricKey.ACTOR_INVOCATION_COUNT,
+            count,
+            plan_id,
+            labels=dict(labels),
+        )
+
+    @staticmethod
+    def actor_latency(plan_id: str, latency_ms: float, **labels: Any) -> MetricEntry:
+        """Record ``ACTOR_LATENCY_MS``."""
+        return MetricCollector.record(
+            OperationalMetricKey.ACTOR_LATENCY_MS,
+            latency_ms,
+            plan_id,
+            labels=dict(labels),
+        )
+
+    @staticmethod
+    def tool_invocation_count(plan_id: str, count: float, **labels: Any) -> MetricEntry:
+        """Record ``TOOL_INVOCATION_COUNT``."""
+        return MetricCollector.record(
+            OperationalMetricKey.TOOL_INVOCATION_COUNT,
+            count,
+            plan_id,
+            labels=dict(labels),
+        )
+
+    @staticmethod
+    def tool_error_rate(plan_id: str, rate: float, **labels: Any) -> MetricEntry:
+        """Record ``TOOL_ERROR_RATE``."""
+        return MetricCollector.record(
+            OperationalMetricKey.TOOL_ERROR_RATE,
+            rate,
+            plan_id,
+            labels=dict(labels),
+        )
+
+    @staticmethod
+    def context_build_time(
+        plan_id: str, duration_ms: float, **labels: Any
+    ) -> MetricEntry:
+        """Record ``CONTEXT_BUILD_TIME_MS``."""
+        return MetricCollector.record(
+            OperationalMetricKey.CONTEXT_BUILD_TIME_MS,
+            duration_ms,
+            plan_id,
+            labels=dict(labels),
+        )
+
+    @staticmethod
+    def context_token_count(plan_id: str, count: float, **labels: Any) -> MetricEntry:
+        """Record ``CONTEXT_TOKEN_COUNT``."""
+        return MetricCollector.record(
+            OperationalMetricKey.CONTEXT_TOKEN_COUNT,
+            count,
+            plan_id,
+            labels=dict(labels),
+        )
+
     @staticmethod
     def llm_call_count(plan_id: str, count: float, **labels: Any) -> MetricEntry:
         """Record ``LLM_CALL_COUNT``."""
@@ -172,3 +458,33 @@ class MetricCollector:
             plan_id,
             labels=dict(labels),
         )
+
+    @staticmethod
+    def llm_total_tokens(plan_id: str, count: float, **labels: Any) -> MetricEntry:
+        """Record ``LLM_TOTAL_TOKENS``."""
+        return MetricCollector.record(
+            OperationalMetricKey.LLM_TOTAL_TOKENS,
+            count,
+            plan_id,
+            labels=dict(labels),
+        )
+
+    @staticmethod
+    def llm_total_cost(plan_id: str, cost_usd: float, **labels: Any) -> MetricEntry:
+        """Record ``LLM_TOTAL_COST_USD``."""
+        return MetricCollector.record(
+            OperationalMetricKey.LLM_TOTAL_COST_USD,
+            cost_usd,
+            plan_id,
+            labels=dict(labels),
+        )
+
+    @staticmethod
+    def llm_avg_latency(plan_id: str, latency_ms: float, **labels: Any) -> MetricEntry:
+        """Record ``LLM_AVG_LATENCY_MS``."""
+        return MetricCollector.record(
+            OperationalMetricKey.LLM_AVG_LATENCY_MS,
+            latency_ms,
+            plan_id,
+            labels=dict(labels),
+        )

src/cleveragents/infrastructure/observability/__init__.py

@@ -0,0 +1,13 @@
+"""Observability infrastructure for metric emission.
+
+Provides ``MetricsEmitter`` for structured log emission of
+``MetricEntry`` objects in local mode.
+
+Based on Forgejo issue #579.
+"""
+
+from __future__ import annotations
+
+from cleveragents.infrastructure.observability.metrics_emitter import MetricsEmitter
+
+__all__ = ["MetricsEmitter"]

src/cleveragents/infrastructure/observability/metrics_emitter.py

@@ -0,0 +1,115 @@
+"""Metrics emitter for structured log emission.
+
+``MetricsEmitter`` accepts ``MetricEntry`` objects and emits them as
+structured log entries via ``structlog``.  Local-mode emission is the
+default; Prometheus export is a future extension gated behind a
+configuration flag.
+
+Based on Forgejo issue #579.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+import structlog
+
+if TYPE_CHECKING:
+    from cleveragents.config.settings import Settings
+    from cleveragents.domain.models.observability.metrics import MetricEntry
+
+logger = structlog.get_logger(__name__)
+
+
+class MetricsEmitter:
+    """Emits ``MetricEntry`` objects as structured log entries.
+
+    Attributes:
+        _enabled: Whether metric emission is active.
+        _prometheus_enabled: Whether Prometheus export is enabled.
+    """
+
+    def __init__(
+        self,
+        *,
+        enabled: bool = True,
+        prometheus_enabled: bool = False,
+    ) -> None:
+        """Initialise the metrics emitter.
+
+        Args:
+            enabled: Master switch for metric emission.
+            prometheus_enabled: Enable Prometheus export (future).
+        """
+        self._enabled = enabled
+        self._prometheus_enabled = prometheus_enabled
+
+    @classmethod
+    def from_settings(cls, settings: Settings) -> MetricsEmitter:
+        """Create a ``MetricsEmitter`` from application settings.
+
+        Args:
+            settings: Application settings instance.
+
+        Returns:
+            A configured ``MetricsEmitter``.
+        """
+        return cls(
+            enabled=getattr(settings, "metrics_enabled", True),
+            prometheus_enabled=getattr(settings, "metrics_export_prometheus", False),
+        )
+
+    @property
+    def enabled(self) -> bool:
+        """Whether metric emission is active."""
+        return self._enabled
+
+    @enabled.setter
+    def enabled(self, value: bool) -> None:
+        """Set the enabled state."""
+        self._enabled = value
+
+    @property
+    def prometheus_enabled(self) -> bool:
+        """Whether Prometheus export is enabled."""
+        return self._prometheus_enabled
+
+    def emit(self, entry: MetricEntry) -> None:
+        """Emit a single metric entry as a structured log event.
+
+        When ``enabled`` is ``False``, this is a no-op.
+
+        Args:
+            entry: The metric entry to emit.
+        """
+        if not self._enabled:
+            return
+
+        log_data: dict[str, Any] = {
+            "metric_key": str(entry.key),
+            "metric_value": entry.value,
+            "plan_id": entry.plan_id,
+            "timestamp": entry.timestamp.isoformat(),
+        }
+        if entry.metric_type is not None:
+            log_data["metric_type"] = str(entry.metric_type)
+        if entry.labels:
+            log_data["labels"] = entry.labels
+
+        logger.info("metric.recorded", **log_data)
+
+    def emit_batch(self, entries: list[MetricEntry]) -> int:
+        """Emit multiple metric entries.
+
+        Args:
+            entries: List of metric entries to emit.
+
+        Returns:
+            Number of entries emitted (0 when disabled).
+        """
+        if not self._enabled:
+            return 0
+
+        for entry in entries:
+            self.emit(entry)
+        return len(entries)