feat(sec-core): cosh hook for security observability#528
Conversation
RemindD
force-pushed
the
feature/sec-core/coshob
branch
from
c88c3e6 to
7fb8ead
Compare
|
|
||
|
|
||
| def _record_observability(record: dict[str, Any]) -> None: | ||
| subprocess.run( |
There was a problem hiding this comment.
[NIT] 在 BeforeModel / AfterModel / PreToolUse / PostToolUse / PostToolUseFailure / Stop / UserPromptSubmit 七个事件都会同步 fork 一个 python3 → agent-sec-cli observability record 子进程（CLI 是 Python 入口）会有性能问题，后续需要考虑优化
| metadata = { | ||
| "sessionId": _string_or_empty(input_data.get("session_id")), | ||
| "runId": _string_or_empty(input_data.get("run_id")) | ||
| or _synthetic_id("run", input_data), |
There was a problem hiding this comment.
runID 来自 sha256(_json_dumps(input_data))[:16]，再被 _metadata 用作 runId 的 fallback，同一 run 的 UserPromptSubmit、BeforeModel、PreToolUse、Stop payload 内容各不相同，因此每个事件会得到一个不同的 synthetic-run-XXX，这样能够满足 session 关联的要求吗？
There was a problem hiding this comment.
cosh 的runId是通过prompt id 继承的，格式为{sessionId}##..##{turnid}, e.g. b56f109e-c704-49be-bdce-d7e003d34445########0. 在每一次prompt 对话中都是保持一致的。唯一的例外的UserPromptSubmit的runId，因为hook 运行时runid未赋值，会找到legacy的值。在issue #531 中追踪。
| if not isinstance(llm_response, dict): | ||
| llm_response = {} | ||
|
|
||
| metrics: dict[str, Any] = {"outcome": "success"} |
There was a problem hiding this comment.
目前cosh 侧只有在LLM call 成功时才会进AfterModel hook。如果LLM call 失败了，会被cosh core catch，失败的原因并没有暴露给任何cosh hook
LLM call 失败后的主路径是：
用户消息
-> BeforeModel hook
-> 调 LLM / 读 stream
-> 失败抛错
-> turn.ts catch
-> yield GeminiEventType.Error
-> client.ts 收到 Error 后 return
结果是：
- 不会走 AfterModel
- 不会自动进入 tool call 阶段
- 不会有正常 assistant reply
- 不会触发 Stop hook
- 用户可以之后再发新消息，但这是下一轮，不是当前失败 LLM call 的后续
|
|
||
|
|
||
| def _build_post_tool_use(input_data: dict[str, Any]) -> dict[str, Any] | None: | ||
| metrics: dict[str, Any] = {"status": "success"} |
There was a problem hiding this comment.
after tool call hook 分为post_tool_use和post_tool_use_failure，所以post_tool_use的hook 中的status 硬编码成了success。对于post_tool_use_failure，会区分是interrupted 还是error
def _build_post_tool_use_failure(input_data: dict[str, Any]) -> dict[str, Any] | None:
metrics: dict[str, Any] = {
"status": "interrupted" if input_data.get("is_interrupt") is True else "error"
}
RemindD
force-pushed
the
feature/sec-core/coshob
branch
from
7fb8ead to
af76bf4
Compare
2 participants
Description
Register a cosh extension hook that records prompt, model, tool, and stop events through
agent-sec-cli observability record. The hook maps available hook payload fields into observability metrics, fills stable metadata with synthetic IDs when needed, and always emits an empty hook output so it does not affect run decisions.Add unit coverage for event mapping, payload sizing, synthetic metadata, invalid input handling, no-op output, and best-effort CLI invocation.
Related Issue
closes #
Type of Change
Scope
cosh(copilot-shell)sec-core(agent-sec-core)skill(os-skills)sight(agentsight)tokenless(tokenless)Checklist
cosh: Lint passes, type check passes, and tests passsec-core(Rust):cargo clippy -- -D warningsandcargo fmt --checkpasssec-core(Python): Ruff format and pytest passskill: Skill directory structure is valid and shell scripts pass syntax checksight:cargo clippy -- -D warningsandcargo fmt --checkpasstokenless:cargo clippy -- -D warningsandcargo fmt --checkpasspackage-lock.json/Cargo.lock)Testing
Additional Notes