Files
DECNET/decnet/profiler/behave_shell/_features/cognitive.py
anti 3fc6ea5f75 feat(profiler/behave_shell): emit cognitive.command_branch_diversity
BEHAVE-EXTRACTOR.md Phase A Step 6. Content-based playbook-vs-
adaptive split. Splits CLAUDE-FF (linear_playbook, ~10 distinct
tools) from CLAUDE-CL (adaptive_branching, 5-6 tools with curl
re-invoked) per the 2026-05-02 empirical anchor.

* _features/cognitive.py:command_branch_diversity(ctx) emits one
  Observation in {linear_playbook, adaptive_branching, unknown}.
* unique_first_token_hashes / total_commands ratio. ≥ 0.80 →
  linear_playbook, otherwise adaptive_branching (the doc instructs
  bias-to-adaptive in the middle band — that's the discriminative
  signal we actually want).
* < 5 commands → "unknown" at confidence 1.0 (the absence of data
  is itself a high-confidence answer per the registry's allowed
  vocabulary). Zero-command session skips emission entirely.

Tests cover unique-tokens → linear, repeated-tokens → adaptive,
middle band → adaptive (bias), under-floor → unknown @ 1.0, plus
PII regression: raw tokens never appear in the serialised
observation.
2026-05-03 07:54:13 -04:00

103 lines
3.4 KiB
Python

"""``cognitive.*`` feature functions.
Step 5: ``cognitive.inter_command_latency_class``.
Step 6: ``cognitive.command_branch_diversity``.
Step 7: ``cognitive.feedback_loop_engagement``.
Step 8: ``cognitive.inter_command_consistency``.
"""
from __future__ import annotations
import statistics
from typing import Iterator
from decnet_behave_core.spec.envelope import Observation
from decnet.profiler.behave_shell._ctx import SessionContext
from decnet.profiler.behave_shell._features._emit import make_observation
from decnet.profiler.behave_shell._thresholds import (
BRANCH_DIVERSITY_LINEAR_MIN,
INTER_CMD_DELIBERATE_MAX,
INTER_CMD_INSTANT_MAX,
INTER_CMD_LLM_HEAVYWEIGHT_MAX,
INTER_CMD_LLM_LIGHTWEIGHT_MAX,
INTER_CMD_TYPING_MAX,
MIN_COMMANDS_FOR_FULL_CONFIDENCE,
)
def _bucket_inter_cmd_latency(median_iat: float) -> str:
if median_iat <= INTER_CMD_INSTANT_MAX:
return "instant"
if median_iat <= INTER_CMD_TYPING_MAX:
return "typing_speed"
if median_iat <= INTER_CMD_DELIBERATE_MAX:
return "deliberate"
if median_iat <= INTER_CMD_LLM_LIGHTWEIGHT_MAX:
return "llm_lightweight"
if median_iat <= INTER_CMD_LLM_HEAVYWEIGHT_MAX:
return "llm_heavyweight"
return "long"
def inter_command_latency_class(ctx: SessionContext) -> Iterator[Observation]:
"""Emit ``cognitive.inter_command_latency_class``.
Operator's *thinking pace* between commands, bucketed against
calibrated thresholds. Splits LW-sim / CLAUDE-FF / CLAUDE-CL.
"""
if not ctx.inter_cmd_iats:
return
median_iat = statistics.median(ctx.inter_cmd_iats)
bucket = _bucket_inter_cmd_latency(median_iat)
# Sample-size honesty: < 5 commands → halve confidence
if len(ctx.commands) < MIN_COMMANDS_FOR_FULL_CONFIDENCE:
confidence = 0.40
else:
confidence = 0.80
yield make_observation(
ctx,
primitive="cognitive.inter_command_latency_class",
value=bucket,
confidence=confidence,
)
def command_branch_diversity(ctx: SessionContext) -> Iterator[Observation]:
"""Emit ``cognitive.command_branch_diversity``.
Content-based discriminator (no timing): unique first-token ratio
over total commands. Splits CLAUDE-FF (linear_playbook) from
CLAUDE-CL (adaptive_branching). The empirical anchor on
2026-05-02: fire-and-forget runs ~10 distinct tools; closed-loop
runs 5-6 with ``curl`` re-invoked as the operator chases threads.
"""
n = len(ctx.commands)
if n == 0:
# No commands at all → nothing honest to say. Skip emission.
return
if n < MIN_COMMANDS_FOR_FULL_CONFIDENCE:
# Registry admits "unknown"; absence of *enough* data is itself
# a high-confidence answer.
yield make_observation(
ctx,
primitive="cognitive.command_branch_diversity",
value="unknown",
confidence=1.0,
)
return
unique = len({c.first_token_hash for c in ctx.commands})
ratio = unique / n
if ratio >= BRANCH_DIVERSITY_LINEAR_MIN:
value = "linear_playbook"
else:
# Anything below the linear floor is treated as adaptive — the
# operator is reusing tools, the discriminative signal we
# actually want.
value = "adaptive_branching"
yield make_observation(
ctx,
primitive="cognitive.command_branch_diversity",
value=value,
confidence=0.80,
)