一、通用（Python 脚本）：
Claude Code 的设置文件 ~\.claude\settings.json 增加钩子（根据实际情况，文件路径需要修改）：
{
"hooks": {
"SessionEnd": [
{
"matcher": ".*",
"hooks": [
{
"type": "command",
"command": "python3 /root/claude/hooks/audit.py SessionEnd",
"timeout": 15
}
]
}
],
"UserPromptSubmit": [
{
"hooks": [
{
"type": "command",
"command": "python3 /root/claude/hooks/audit.py UserPromptSubmit",
"timeout": 10
}
]
}
],
"PreToolUse": [
{
"matcher": ".*",
"hooks": [
{
"type": "command",
"command": "python3 /root/claude/hooks/audit.py PreToolUse",
"timeout": 10
}
]
}
],
"PostToolUse": [
{
"matcher": ".*",
"hooks": [
{
"type": "command",
"command": "python3 /root/claude/hooks/audit.py PostToolUse",
"timeout": 10
}
]
}
],
"PermissionRequest": [
{
"matcher": ".*",
"hooks": [
{
"type": "command",
"command": "python3 /root/claude/hooks/audit.py PermissionRequest",
"timeout": 10
}
]
}
],
"Stop": [
{
"hooks": [
{
"type": "command",
"command": "python3 /root/claude/hooks/audit.py Stop",
"timeout": 15
}
]
}
]
}
}
在这个路径下增加执行命令文件 /root/claude/hooks/audit.py：
# This code is supported by the website: https://www.guanjihuan.com
# The newest version of this code is on the web page: https://www.guanjihuan.com/archives/49203
#!/usr/bin/env python3
"""Audit hook for Claude Code — Linux port of audit.ps1.
Reads an event name as argv[1] and a JSON payload on stdin, then mutates a
daily JSON log under $CLAUDE_AUDIT_LOG_DIR (default ~/claude/logs). Cross-process
safety is provided by an fcntl flock on a per-log-dir lock file (replaces the
Windows named mutex in the original). All event handlers and the StatusLine
behaviour are kept 1:1 with the PowerShell source.
"""
import fcntl
import json
import os
import re
import sys
import time
import uuid
from datetime import datetime, timedelta
from pathlib import Path
# --- Config ---
LOG_DIR = Path(os.environ.get(
'CLAUDE_AUDIT_LOG_DIR',
os.path.expanduser('~/claude/logs'),
))
LOG_FILE = LOG_DIR / f"{datetime.now():%Y-%m-%d}.json"
ERROR_LOG = LOG_DIR / 'audit-error.log'
MUTEX_TIMEOUT_MS = 5000
STATUS_RUNNING = 'running'
STATUS_AWAITING_USER = 'awaiting_user'
STATUS_COMPLETED = 'completed'
STATUS_STOPPED = 'stopped'
IN_PROGRESS_STATUSES = (STATUS_RUNNING, STATUS_AWAITING_USER)
KNOWN_EVENTS = (
'SessionEnd', 'UserPromptSubmit', 'PreToolUse',
'PostToolUse', 'PermissionRequest', 'Stop', 'StatusLine',
)
# Per-invocation cache for transcript model lookup.
_model_cache_path = None
_model_cache_mtime = None
_model_cache_value = None
# --- Error / IO helpers ---
def write_err(msg, stage):
"""Append a timestamped line to the error log; never raise."""
try:
line = f"{datetime.now().isoformat()} [{stage}] {msg}"
with open(ERROR_LOG, 'a', encoding='utf-8') as f:
f.write(line + '\n')
except Exception:
pass
def read_stdin_json():
"""Read and parse JSON from stdin. Returns None on empty/invalid input."""
try:
raw = sys.stdin.read()
if raw and raw.strip():
return json.loads(raw)
except Exception as e:
write_err(f'stdin_parse_failed: {e}', 'parse')
return None
def get_property(obj, name):
"""dict.get() / getattr() shim matching the PSObject property accessor."""
if obj is None:
return None
if isinstance(obj, dict):
return obj.get(name)
return getattr(obj, name, None)
def write_status_line(line):
sys.stdout.write(line + '\n')
sys.stdout.flush()
# --- Model helpers ---
def test_concrete_model(model):
"""A model name is 'concrete' if it's a non-empty string not in the
placeholder set (auto/default/unknown/null)."""
if not model or not isinstance(model, str) or not model.strip():
return False
return model.strip().lower() not in ('auto', 'default', 'unknown', 'null')
def get_model_name(value):
"""Pull a model identifier out of either a string or an object with one
of the well-known field names. Returns the first concrete match, falling
back to the first non-empty text or str(value)."""
if value is None:
return None
if isinstance(value, str):
return value.strip()
fallback = None
if isinstance(value, dict):
for name in ('id', 'model', 'name', 'display_name'):
model = get_property(value, name)
if not model:
continue
text = str(model).strip()
if not fallback:
fallback = text
if test_concrete_model(text):
return text
if fallback:
return fallback
return str(value).strip()
def get_model_from_payload(payload):
if not payload:
return None
model = get_model_name(get_property(payload, 'model'))
if test_concrete_model(model):
return model
return None
def get_model_from_environment():
for name in ('ANTHROPIC_MODEL', 'CLAUDE_MODEL', 'ANTHROPIC_SMALL_FAST_MODEL'):
model = os.environ.get(name)
if test_concrete_model(model):
return model.strip()
return None
def get_model_from_transcript(path):
"""Walk a Claude Code transcript JSONL file backwards to find the most
recent assistant message that carries a concrete model name. Cached by
(path, mtime) for the lifetime of this invocation."""
global _model_cache_path, _model_cache_mtime, _model_cache_value
if not path:
return None
p = Path(path)
if not p.exists():
return None
try:
mtime = p.stat().st_mtime
if _model_cache_path == str(p) and _model_cache_mtime == mtime:
return _model_cache_value
with open(p, 'r', encoding='utf-8') as f:
raw = f.read()
if not raw.strip():
return None
for line in reversed(raw.splitlines()):
if not line.strip():
continue
try:
obj = json.loads(line)
except Exception:
continue
if not isinstance(obj, dict):
continue
if obj.get('type') != 'assistant':
continue
msg = obj.get('message')
if isinstance(msg, dict) and msg.get('model'):
_model_cache_path = str(p)
_model_cache_mtime = mtime
_model_cache_value = msg['model']
return _model_cache_value
except Exception as e:
write_err(f'transcript_model_read_failed: {e}', 'transcript')
return None
def get_model_from_payload_or_transcript(payload):
if payload:
transcript_path = get_property(payload, 'transcript_path')
if transcript_path:
model = get_model_from_transcript(str(transcript_path))
if test_concrete_model(model):
return model
model = get_model_from_payload(payload)
if test_concrete_model(model):
return model
return get_model_from_environment()
def update_prompt_model(prompt, payload):
"""If the prompt has no concrete model, fill it in from payload/transcript/
env. Mutates prompt in place. Returns True iff it changed."""
if not prompt:
return False
if test_concrete_model(str(prompt.get('model'))):
return False
model = get_model_from_payload_or_transcript(payload)
if test_concrete_model(model):
prompt['model'] = model
return True
return False
# --- Log IO ---
def read_log(path=None):
"""Read the log file and return its sessions as a list of dicts."""
p = Path(path) if path else LOG_FILE
if not p.exists():
return []
try:
with open(p, 'r', encoding='utf-8') as f:
raw = f.read()
if not raw.strip():
return []
data = json.loads(raw)
if data is None:
return []
if isinstance(data, list):
return data
return [data]
except Exception as e:
write_err(f'log_read_failed: {e}', 'read')
return []
def write_log(sessions, path=None):
"""Atomically write the sessions array as pretty JSON. Validates the
output by re-parsing before swapping the temp file into place."""
p = Path(path) if path else LOG_FILE
tmp = p.with_name(f'{p.name}.tmp.{os.getpid()}')
try:
items = list(sessions) if sessions else []
if not items:
json_text = '[]'
else:
json_items = [json.dumps(s, ensure_ascii=False, indent=2) for s in items]
json_text = '[\n' + ',\n'.join(json_items) + '\n]'
with open(tmp, 'w', encoding='utf-8') as f:
f.write(json_text)
with open(tmp, 'r', encoding='utf-8') as f:
json.loads(f.read())
os.replace(tmp, p)
return True
except Exception as e:
write_err(f'log_write_failed: {e}', 'write')
try:
tmp.unlink()
except Exception:
pass
return False
# --- Session / prompt state ---
def find_session(sessions, sid):
if not sid:
return None
for s in sessions:
if str(s.get('session_id')) == sid:
return s
return None
def find_last_in_progress(sessions, sid):
"""Last prompt (by list order) whose status is still running or
awaiting_user. Mirrors the original's 'keep overwriting $last' loop."""
session = find_session(sessions, sid)
if not session or not session.get('prompts'):
return None
last = None
for p in session['prompts']:
if str(p.get('status')) in IN_PROGRESS_STATUSES:
last = p
return last
def complete_prompt(prompt, now):
if not prompt or str(prompt.get('status')) not in IN_PROGRESS_STATUSES:
return False
prompt['status'] = STATUS_COMPLETED
prompt['end_time'] = now
prompt['duration'] = get_duration_hms(str(prompt.get('start_time')), now)
return True
def stop_prompt(prompt, now):
if not prompt or str(prompt.get('status')) not in IN_PROGRESS_STATUSES:
return False
prompt['status'] = STATUS_STOPPED
prompt['end_time'] = now
prompt['duration'] = get_duration_hms(str(prompt.get('start_time')), now)
return True
def invoke_yesterday_cleanup():
"""Mark any in-progress prompts in yesterday's log as stopped with
end_time = 23:59:59. Idempotent: stop_prompt returns False on done prompts."""
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
yesterday_file = LOG_DIR / f'{yesterday}.json'
if not yesterday_file.exists():
return
try:
y_sessions = read_log(yesterday_file)
convert_prompt_lists(y_sessions)
y_end = f'{yesterday} 23:59:59'
cleanup_changed = False
for s in y_sessions:
for p in s.get('prompts') or []:
if stop_prompt(p, y_end):
cleanup_changed = True
if cleanup_changed:
write_log(y_sessions, yesterday_file)
except Exception as e:
write_err(f'yesterday_cleanup_failed: {e}', 'cleanup')
def get_duration_hms(start, end):
"""Format a duration as HH:MM:SS, floor-zero on negative deltas."""
if not start or not end:
return None
try:
fmt = '%Y-%m-%d %H:%M:%S'
s = datetime.strptime(start, fmt)
e = datetime.strptime(end, fmt)
delta = e - s
if delta.total_seconds() < 0:
return '00:00:00'
total_seconds = int(delta.total_seconds())
h = total_seconds // 3600
m = (total_seconds % 3600) // 60
sec = total_seconds % 60
return f'{h:02d}:{m:02d}:{sec:02d}'
except Exception as e:
write_err(f"duration_parse_failed start='{start}' end='{end}': {e}", 'duration')
return None
def truncate(text, length):
if not text or len(text) <= length:
return text
return text[:length]
def get_short_model(model):
"""Compact model label: strip 'claude-' prefix, join first 3 dash/slash
parts, cap at 12 chars."""
if not model or not isinstance(model, str) or not model.strip():
return ''
short = re.sub(r'^claude-', '', model.strip())
parts = re.split(r'[-/]', short)
short = '-'.join(parts[:3])
if len(short) > 12:
short = short[:12]
return short
def convert_prompt_lists(sessions):
"""Ensure each session has a real list at .prompts (always true for
json.loads output, but kept for parity with the PS version)."""
for s in sessions:
if not isinstance(s, dict):
continue
prompts = []
if s.get('prompts'):
for p in s['prompts']:
prompts.append(p)
s['prompts'] = prompts
# --- Main ---
def acquire_lock(lock_path, timeout_ms):
"""Try to acquire an exclusive flock within timeout_ms. Returns
(fd, acquired). fd is left open on success so the lock stays held."""
fd = open(lock_path, 'w')
deadline = time.monotonic() + timeout_ms / 1000.0
while True:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
return fd, True
except (BlockingIOError, OSError):
if time.monotonic() >= deadline:
return fd, False
time.sleep(0.05)
def release_lock(fd):
if fd is None:
return
try:
fcntl.flock(fd, fcntl.LOCK_UN)
except Exception:
pass
try:
fd.close()
except Exception:
pass
def handle_status_line(payload):
try:
sessions = read_log()
sid = str(get_property(payload, 'session_id')) if payload else None
session = find_session(sessions, sid) if sid else None
if not session and sessions:
for i in range(len(sessions) - 1, -1, -1):
if find_last_in_progress(sessions, str(sessions[i].get('session_id'))):
session = sessions[i]
break
if not session and sessions:
for i in range(len(sessions) - 1, -1, -1):
if sessions[i].get('prompts'):
session = sessions[i]
break
if not session:
write_status_line('· idle')
return
prompt = find_last_in_progress(sessions, str(session.get('session_id')))
if not prompt and session.get('prompts'):
prompt = session['prompts'][-1]
if not prompt:
write_status_line('· idle')
return
status = str(prompt.get('status'))
if status == STATUS_RUNNING:
glyph = '*'
elif status == STATUS_AWAITING_USER:
glyph = '?'
else:
glyph = '·'
prompt_model = str(prompt.get('model'))
display_model = prompt_model if test_concrete_model(prompt_model) else get_model_from_environment()
model_short = get_short_model(display_model)
content = truncate(re.sub(r'[\r\n]+', ' ', str(prompt.get('content', ''))), 30)
line = f"{glyph} {status}"
if model_short:
line += f" {model_short}"
if content:
line += f" {content}"
write_status_line(line)
except Exception as e:
write_err(f'status_failed: {e}', 'status')
write_status_line('· ?')
def handle_event(event_type, payload):
sessions = read_log()
convert_prompt_lists(sessions)
sid = str(get_property(payload, 'session_id')) if payload else None
if not sid or not sid.strip():
sid = str(uuid.uuid4())
current_session = find_session(sessions, sid)
if not current_session and event_type == 'UserPromptSubmit':
# Lazy day-rollover cleanup: only when we're about to create today's
# log file (first UserPromptSubmit of the day, new session).
today = datetime.now().strftime('%Y-%m-%d')
today_file = LOG_DIR / f'{today}.json'
if not today_file.exists():
invoke_yesterday_cleanup()
project = str(get_property(payload, 'cwd')) if payload else None
current_session = {
'project': project,
'session_id': sid,
'prompts': [],
}
sessions.append(current_session)
changed = False
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
if event_type == 'UserPromptSubmit':
content = ''
if payload:
for key in ('prompt', 'user_prompt', 'message'):
value = get_property(payload, key)
if value is not None:
content = str(value)
break
content = truncate(content, 500)
for prompt in current_session.get('prompts') or []:
if stop_prompt(prompt, now):
changed = True
model = get_model_from_payload_or_transcript(payload)
current_session['prompts'].append({
'content': content,
'model': model,
'start_time': now,
'end_time': '',
'duration': None,
'status': STATUS_RUNNING,
})
changed = True
elif event_type == 'PreToolUse':
prompt = find_last_in_progress(sessions, sid)
if prompt:
tool_name = str(get_property(payload, 'tool_name')) if payload else ''
if tool_name == 'AskUserQuestion':
if str(prompt.get('status')) != STATUS_AWAITING_USER:
prompt['status'] = STATUS_AWAITING_USER
changed = True
elif str(prompt.get('status')) == STATUS_AWAITING_USER:
prompt['status'] = STATUS_RUNNING
changed = True
if update_prompt_model(prompt, payload):
changed = True
elif event_type == 'PermissionRequest':
prompt = find_last_in_progress(sessions, sid)
if prompt:
if str(prompt.get('status')) != STATUS_AWAITING_USER:
prompt['status'] = STATUS_AWAITING_USER
changed = True
if update_prompt_model(prompt, payload):
changed = True
elif event_type == 'PostToolUse':
prompt = find_last_in_progress(sessions, sid)
if prompt:
if str(prompt.get('status')) == STATUS_AWAITING_USER:
prompt['status'] = STATUS_RUNNING
changed = True
if update_prompt_model(prompt, payload):
changed = True
elif event_type == 'Stop':
prompt = find_last_in_progress(sessions, sid)
if prompt:
if update_prompt_model(prompt, payload):
changed = True
if complete_prompt(prompt, now):
changed = True
elif event_type == 'SessionEnd':
prompt = find_last_in_progress(sessions, sid)
if prompt:
if update_prompt_model(prompt, payload):
changed = True
if stop_prompt(prompt, now):
changed = True
if changed:
write_log(sessions)
def main():
if len(sys.argv) < 2:
sys.exit(0)
event_type = sys.argv[1]
if event_type not in KNOWN_EVENTS:
sys.exit(0)
try:
LOG_DIR.mkdir(parents=True, exist_ok=True)
except Exception as e:
write_err(f'logdir_create_failed: {e}', 'init')
payload = read_stdin_json()
if event_type == 'StatusLine':
handle_status_line(payload)
sys.exit(0)
lock_fd, acquired = None, False
try:
lock_fd, acquired = acquire_lock(LOG_DIR / '.audit.lock', MUTEX_TIMEOUT_MS)
except Exception as e:
write_err(f'mutex_wait_failed: {e}', 'mutex')
if not acquired:
write_err(f'mutex_timeout ({MUTEX_TIMEOUT_MS}ms) event={event_type}', 'mutex')
release_lock(lock_fd)
sys.exit(0)
try:
handle_event(event_type, payload)
except Exception as e:
write_err(f'handler_failed: {e}', 'handler')
finally:
release_lock(lock_fd)
sys.exit(0)
if __name__ == '__main__':
main()
二、Windows 版本（PowerShell 脚本）：
Claude Code 的设置文件 C:\Users\guan\.claude\settings.json 增加钩子：
{
"hooks": {
"SessionEnd": [
{
"matcher": ".*",
"hooks": [
{
"type": "command",
"command": "powershell.exe -NoProfile -ExecutionPolicy Bypass -File \"d:/claude/hooks/audit.ps1\" SessionEnd",
"timeout": 15
}
]
}
],
"UserPromptSubmit": [
{
"hooks": [
{
"type": "command",
"command": "powershell.exe -NoProfile -ExecutionPolicy Bypass -File \"d:/claude/hooks/audit.ps1\" UserPromptSubmit",
"timeout": 10
}
]
}
],
"PreToolUse": [
{
"matcher": ".*",
"hooks": [
{
"type": "command",
"command": "powershell.exe -NoProfile -ExecutionPolicy Bypass -File \"d:/claude/hooks/audit.ps1\" PreToolUse",
"timeout": 10
}
]
}
],
"PostToolUse": [
{
"matcher": ".*",
"hooks": [
{
"type": "command",
"command": "powershell.exe -NoProfile -ExecutionPolicy Bypass -File \"d:/claude/hooks/audit.ps1\" PostToolUse",
"timeout": 10
}
]
}
],
"PermissionRequest": [
{
"matcher": ".*",
"hooks": [
{
"type": "command",
"command": "powershell.exe -NoProfile -ExecutionPolicy Bypass -File \"d:/claude/hooks/audit.ps1\" PermissionRequest",
"timeout": 10
}
]
}
],
"Stop": [
{
"hooks": [
{
"type": "command",
"command": "powershell.exe -NoProfile -ExecutionPolicy Bypass -File \"d:/claude/hooks/audit.ps1\" Stop",
"timeout": 15
}
]
}
]
}
}
在这个路径下增加执行命令文件 d:/claude/hooks/audit.ps1：
# This code is supported by the website: https://www.guanjihuan.com
# The newest version of this code is on the web page: https://www.guanjihuan.com/archives/49203
[CmdletBinding()]
param([Parameter(Mandatory, Position=0)][string]$EventType)
$Utf8NoBom = [System.Text.UTF8Encoding]::new($false)
[Console]::InputEncoding = $Utf8NoBom
[Console]::OutputEncoding = $Utf8NoBom
$LogDir = 'D:\claude\logs'
$LogFile = Join-Path $LogDir "$(Get-Date -Format 'yyyy-MM-dd').json"
$ErrorLog = Join-Path $LogDir 'audit-error.log'
$MutexTimeoutMs = 5000
$StatusRunning = 'running'
$StatusAwaitingUser = 'awaiting_user'
$StatusCompleted = 'completed'
$StatusStopped = 'stopped'
$InProgressStatuses = @($StatusRunning, $StatusAwaitingUser)
New-Item -ItemType Directory -Path $LogDir -Force -ErrorAction SilentlyContinue | Out-Null
function Write-Err([string]$Msg, [string]$Stage) {
try {
$line = "$(Get-Date -Format 'o') [$Stage] $Msg"
[System.IO.File]::AppendAllText($ErrorLog, $line + "`n", $Utf8NoBom)
} catch { }
}
function Read-StdinJson {
try {
$reader = [System.IO.StreamReader]::new([Console]::OpenStandardInput(), $Utf8NoBom)
$stdin = $reader.ReadToEnd()
if ($stdin -and -not [string]::IsNullOrWhiteSpace($stdin)) {
return $stdin | ConvertFrom-Json -ErrorAction Stop
}
} catch {
Write-Err "stdin_parse_failed: $_" 'parse'
}
return $null
}
function Get-PropertyValue($Obj, [string]$Name) {
if ($null -eq $Obj) { return $null }
$prop = $Obj.PSObject.Properties[$Name]
if ($prop) { return $prop.Value }
return $null
}
function Test-ConcreteModel([string]$Model) {
if ([string]::IsNullOrWhiteSpace($Model)) { return $false }
return ($Model.Trim().ToLowerInvariant() -notin @('auto', 'default', 'unknown', 'null'))
}
function Get-ModelName($Value) {
if ($null -eq $Value) { return $null }
if ($Value -is [string]) { return $Value.Trim() }
$fallback = $null
foreach ($name in @('id', 'model', 'name', 'display_name')) {
$model = Get-PropertyValue $Value $name
if (-not $model) { continue }
$text = ([string]$model).Trim()
if (-not $fallback) { $fallback = $text }
if (Test-ConcreteModel $text) { return $text }
}
if ($fallback) { return $fallback }
return ([string]$Value).Trim()
}
function Get-ModelFromPayload($Payload) {
if (-not $Payload) { return $null }
$model = Get-ModelName (Get-PropertyValue $Payload 'model')
if (Test-ConcreteModel $model) { return $model }
return $null
}
function Get-ModelFromEnvironment {
foreach ($name in @('ANTHROPIC_MODEL', 'CLAUDE_MODEL', 'ANTHROPIC_SMALL_FAST_MODEL')) {
$model = [Environment]::GetEnvironmentVariable($name)
if (Test-ConcreteModel $model) { return $model.Trim() }
}
return $null
}
function Get-ModelFromPayloadOrTranscript($Payload) {
$model = $null
if ($Payload) {
$transcriptPath = Get-PropertyValue $Payload 'transcript_path'
if ($transcriptPath) {
$model = Get-ModelFromTranscript ([string]$transcriptPath)
if (Test-ConcreteModel $model) { return $model }
}
$model = Get-ModelFromPayload $Payload
if (Test-ConcreteModel $model) { return $model }
}
return (Get-ModelFromEnvironment)
}
function Update-PromptModel($Prompt, $Payload) {
if (-not $Prompt) { return $false }
if (Test-ConcreteModel ([string]$Prompt.model)) { return $false }
$model = Get-ModelFromPayloadOrTranscript $Payload
if (Test-ConcreteModel $model) {
$Prompt.model = $model
return $true
}
return $false
}
function ConvertTo-ReadableJson {
[CmdletBinding()]
param(
[Parameter(ValueFromPipeline)]$InputObject,
[int]$Depth = 32
)
process {
$json = $InputObject | ConvertTo-Json -Depth $Depth
if ($json -match '\\u[0-9A-Fa-f]{4}') {
$json = [regex]::Replace($json, '\\u([0-9A-Fa-f]{4})', {
param($m) [char][Convert]::ToInt32($m.Groups[1].Value, 16)
})
}
return $json
}
}
function Read-Log {
param([string]$Path = $LogFile)
if (-not (Test-Path -LiteralPath $Path)) { return @() }
try {
$raw = [System.IO.File]::ReadAllText($Path, $Utf8NoBom)
if ([string]::IsNullOrWhiteSpace($raw)) { return @() }
$data = $raw | ConvertFrom-Json -ErrorAction Stop
if ($null -eq $data) { return @() }
return @($data)
} catch {
Write-Err "log_read_failed: $_" 'read'
return @()
}
}
function Write-Log {
param(
[Parameter(Mandatory)][array]$Sessions,
[string]$Path = $LogFile
)
$tmp = "$Path.tmp.$PID"
try {
$items = @($Sessions)
if ($items.Count -eq 0) {
$json = '[]'
} else {
$jsonItems = @($items | ForEach-Object { ConvertTo-ReadableJson -InputObject $_ -Depth 32 })
$json = "[`n" + ($jsonItems -join ",`n") + "`n]"
}
[System.IO.File]::WriteAllText($tmp, $json, $Utf8NoBom)
$null = ([System.IO.File]::ReadAllText($tmp, $Utf8NoBom) | ConvertFrom-Json -ErrorAction Stop)
Move-Item -LiteralPath $tmp -Destination $Path -Force
return $true
} catch {
Write-Err "log_write_failed: $_" 'write'
if (Test-Path -LiteralPath $tmp) { Remove-Item -LiteralPath $tmp -Force -ErrorAction SilentlyContinue }
return $false
}
}
function Find-Session([array]$Sessions, [string]$Sid) {
foreach ($s in $Sessions) {
if ([string]$s.session_id -eq $Sid) { return $s }
}
return $null
}
function Find-LastInProgress([array]$Sessions, [string]$Sid) {
$session = Find-Session -Sessions $Sessions -Sid $Sid
if (-not $session -or -not $session.prompts) { return $null }
$last = $null
foreach ($p in @($session.prompts)) {
if ([string]$p.status -in $InProgressStatuses) { $last = $p }
}
return $last
}
function Complete-Prompt($Prompt, [string]$Now) {
if (-not $Prompt -or [string]$Prompt.status -notin $InProgressStatuses) { return $false }
$Prompt.status = $StatusCompleted
$Prompt.end_time = $Now
$Prompt.duration = Get-DurationHms -Start ([string]$Prompt.start_time) -End $Now
return $true
}
function Stop-Prompt($Prompt, [string]$Now) {
if (-not $Prompt -or [string]$Prompt.status -notin $InProgressStatuses) { return $false }
$Prompt.status = $StatusStopped
$Prompt.end_time = $Now
$Prompt.duration = Get-DurationHms -Start ([string]$Prompt.start_time) -End $Now
return $true
}
function Invoke-YesterdayCleanup {
# Marks any in-progress prompts in yesterday's log as stopped with
# end_time = yesterday 23:59:59. Idempotent: re-running on an already
# cleaned-up file is a no-op (Stop-Prompt returns false on done prompts).
$yesterday = (Get-Date).AddDays(-1).ToString('yyyy-MM-dd')
$yesterdayFile = Join-Path $LogDir "$yesterday.json"
if (-not (Test-Path -LiteralPath $yesterdayFile)) { return }
try {
$ySessions = @(Read-Log -Path $yesterdayFile)
Convert-PromptLists -Sessions $ySessions
$yEnd = "$yesterday 23:59:59"
$cleanupChanged = $false
foreach ($s in $ySessions) {
foreach ($p in @($s.prompts)) {
if (Stop-Prompt $p $yEnd) { $cleanupChanged = $true }
}
}
if ($cleanupChanged) {
[void](Write-Log -Sessions $ySessions -Path $yesterdayFile)
}
} catch {
Write-Err "yesterday_cleanup_failed: $_" 'cleanup'
}
}
$script:_modelCachePath = $null
$script:_modelCacheMtime = [datetime]::MinValue
$script:_modelCacheValue = $null
function Get-ModelFromTranscript([string]$Path) {
if ([string]::IsNullOrWhiteSpace($Path) -or -not (Test-Path -LiteralPath $Path)) { return $null }
try {
$mtime = (Get-Item -LiteralPath $Path).LastWriteTimeUtc
if ($script:_modelCachePath -eq $Path -and $script:_modelCacheMtime -eq $mtime) {
return $script:_modelCacheValue
}
$raw = [System.IO.File]::ReadAllText($Path, $Utf8NoBom)
if ([string]::IsNullOrEmpty($raw)) { return $null }
$lines = $raw -split "`r?`n"
for ($i = $lines.Count - 1; $i -ge 0; $i--) {
if ([string]::IsNullOrWhiteSpace($lines[$i])) { continue }
$obj = $lines[$i] | ConvertFrom-Json -ErrorAction SilentlyContinue
if ($obj -and [string]$obj.type -eq 'assistant' -and $obj.message -and $obj.message.model) {
$script:_modelCachePath = $Path
$script:_modelCacheMtime = $mtime
$script:_modelCacheValue = [string]$obj.message.model
return $script:_modelCacheValue
}
}
} catch {
Write-Err "transcript_model_read_failed: $_" 'transcript'
}
return $null
}
function Get-DurationHms([string]$Start, [string]$End) {
if ([string]::IsNullOrWhiteSpace($Start) -or [string]::IsNullOrWhiteSpace($End)) { return $null }
try {
$s = [datetime]::ParseExact($Start, 'yyyy-MM-dd HH:mm:ss', $null)
$e = [datetime]::ParseExact($End, 'yyyy-MM-dd HH:mm:ss', $null)
$ts = $e - $s
if ($ts.TotalSeconds -lt 0) { return '00:00:00' }
$h = [int][Math]::Floor($ts.TotalHours)
return ('{0:D2}:{1:D2}:{2:D2}' -f $h, $ts.Minutes, $ts.Seconds)
} catch {
Write-Err "duration_parse_failed start='$Start' end='$End': $_" 'duration'
return $null
}
}
function Truncate([string]$Text, [int]$Length) {
if ([string]::IsNullOrEmpty($Text) -or $Text.Length -le $Length) { return $Text }
return $Text.Substring(0, $Length)
}
function Get-ShortModel([string]$Model) {
if ([string]::IsNullOrWhiteSpace($Model)) { return '' }
$short = $Model -replace '^claude-', ''
$parts = $short -split '[-/]'
$short = ($parts | Select-Object -First 3) -join '-'
if ($short.Length -gt 12) { $short = $short.Substring(0, 12) }
return $short
}
function Convert-PromptLists([array]$Sessions) {
foreach ($s in $Sessions) {
$prompts = [System.Collections.Generic.List[object]]::new()
if ($s.prompts) {
foreach ($p in @($s.prompts)) { $prompts.Add($p) }
}
if ($s.PSObject.Properties['prompts']) {
$s.prompts = $prompts
} else {
$s | Add-Member -MemberType NoteProperty -Name prompts -Value $prompts
}
}
}
function Write-StatusLine([string]$Line) {
[Console]::Out.WriteLine($Line)
}
$KnownEvents = @('SessionEnd', 'UserPromptSubmit', 'PreToolUse', 'PostToolUse', 'PermissionRequest', 'Stop', 'StatusLine')
if ($EventType -notin $KnownEvents) { exit 0 }
$Payload = Read-StdinJson
if ($EventType -eq 'StatusLine') {
try {
$sessions = @(Read-Log)
$sid = if ($Payload) { [string](Get-PropertyValue $Payload 'session_id') } else { $null }
$session = if ($sid) { Find-Session -Sessions $sessions -Sid $sid } else { $null }
if (-not $session -and $sessions.Count -gt 0) {
for ($i = $sessions.Count - 1; $i -ge 0; $i--) {
if (Find-LastInProgress -Sessions $sessions -Sid ([string]$sessions[$i].session_id)) {
$session = $sessions[$i]
break
}
}
}
if (-not $session -and $sessions.Count -gt 0) {
for ($i = $sessions.Count - 1; $i -ge 0; $i--) {
if ($sessions[$i].prompts -and @($sessions[$i].prompts).Count -gt 0) {
$session = $sessions[$i]
break
}
}
}
if (-not $session) {
Write-StatusLine '· idle'
exit 0
}
$prompt = Find-LastInProgress -Sessions $sessions -Sid ([string]$session.session_id)
if (-not $prompt -and $session.prompts) { $prompt = @($session.prompts) | Select-Object -Last 1 }
if (-not $prompt) {
Write-StatusLine '· idle'
exit 0
}
$status = [string]$prompt.status
$glyph = if ($status -eq $StatusRunning) { '*' } elseif ($status -eq $StatusAwaitingUser) { '?' } else { '·' }
$displayModel = if (Test-ConcreteModel ([string]$prompt.model)) { [string]$prompt.model } else { Get-ModelFromEnvironment }
$modelShort = Get-ShortModel $displayModel
$content = Truncate (([string]$prompt.content) -replace '[\r\n]+', ' ') 30
$line = "$glyph $status"
if ($modelShort) { $line += " $modelShort" }
if ($content) { $line += " $content" }
Write-StatusLine $line
} catch {
Write-Err "status_failed: $_" 'status'
Write-StatusLine '· ?'
}
exit 0
}
$mutexName = 'Global\audit-' + ([System.IO.Path]::GetFullPath($LogFile).ToLowerInvariant() -replace '[^a-z0-9]', '_')
$mutex = $null
$acquired = $false
try {
$mutex = [System.Threading.Mutex]::new($false, $mutexName)
try {
$acquired = $mutex.WaitOne($MutexTimeoutMs)
} catch [System.Threading.AbandonedMutexException] {
$acquired = $true
}
} catch {
Write-Err "mutex_wait_failed: $_" 'mutex'
}
if (-not $acquired) {
Write-Err "mutex_timeout (${MutexTimeoutMs}ms) event=$EventType" 'mutex'
if ($mutex) { try { $mutex.Dispose() } catch { } }
exit 0
}
try {
$sessions = [System.Collections.Generic.List[object]]::new()
foreach ($s in @(Read-Log)) { $sessions.Add($s) }
Convert-PromptLists -Sessions @($sessions)
$sid = if ($Payload) { [string](Get-PropertyValue $Payload 'session_id') } else { $null }
if ([string]::IsNullOrWhiteSpace($sid)) { $sid = [guid]::NewGuid().ToString() }
$currentSession = Find-Session -Sessions @($sessions) -Sid $sid
if (-not $currentSession -and $EventType -eq 'UserPromptSubmit') {
# Lazy day-rollover cleanup: only when we're about to create today's
# log file (first UserPromptSubmit of the day, new session).
$today = (Get-Date -Format 'yyyy-MM-dd')
$todayFile = Join-Path $LogDir "$today.json"
if (-not (Test-Path -LiteralPath $todayFile)) {
Invoke-YesterdayCleanup
}
$project = if ($Payload) { [string](Get-PropertyValue $Payload 'cwd') } else { $null }
$currentSession = [pscustomobject][ordered]@{
project = $project
session_id = $sid
prompts = [System.Collections.Generic.List[object]]::new()
}
$sessions.Add($currentSession)
}
$changed = $false
$now = Get-Date -Format 'yyyy-MM-dd HH:mm:ss'
switch ($EventType) {
'UserPromptSubmit' {
$content = ''
if ($Payload) {
foreach ($key in @('prompt', 'user_prompt', 'message')) {
$value = Get-PropertyValue $Payload $key
if ($null -ne $value) { $content = [string]$value; break }
}
}
$content = Truncate $content 500
foreach ($prompt in @($currentSession.prompts)) {
if (Stop-Prompt $prompt $now) { $changed = $true }
}
$model = Get-ModelFromPayloadOrTranscript $Payload
$currentSession.prompts.Add([pscustomobject][ordered]@{
content = $content
model = $model
start_time = $now
end_time = ''
duration = $null
status = $StatusRunning
})
$changed = $true
}
'PreToolUse' {
$prompt = Find-LastInProgress -Sessions @($sessions) -Sid $sid
if ($prompt) {
$toolName = if ($Payload) { [string](Get-PropertyValue $Payload 'tool_name') } else { '' }
if ($toolName -eq 'AskUserQuestion') {
if ([string]$prompt.status -ne $StatusAwaitingUser) {
$prompt.status = $StatusAwaitingUser
$changed = $true
}
} elseif ([string]$prompt.status -eq $StatusAwaitingUser) {
$prompt.status = $StatusRunning
$changed = $true
}
if (Update-PromptModel $prompt $Payload) { $changed = $true }
}
}
'PermissionRequest' {
$prompt = Find-LastInProgress -Sessions @($sessions) -Sid $sid
if ($prompt) {
if ([string]$prompt.status -ne $StatusAwaitingUser) {
$prompt.status = $StatusAwaitingUser
$changed = $true
}
if (Update-PromptModel $prompt $Payload) { $changed = $true }
}
}
'PostToolUse' {
$prompt = Find-LastInProgress -Sessions @($sessions) -Sid $sid
if ($prompt) {
if ([string]$prompt.status -eq $StatusAwaitingUser) {
$prompt.status = $StatusRunning
$changed = $true
}
if (Update-PromptModel $prompt $Payload) { $changed = $true }
}
}
'Stop' {
$prompt = Find-LastInProgress -Sessions @($sessions) -Sid $sid
if ($prompt) {
if (Update-PromptModel $prompt $Payload) { $changed = $true }
if (Complete-Prompt $prompt $now) { $changed = $true }
}
}
'SessionEnd' {
$prompt = Find-LastInProgress -Sessions @($sessions) -Sid $sid
if ($prompt) {
if (Update-PromptModel $prompt $Payload) { $changed = $true }
if (Stop-Prompt $prompt $now) { $changed = $true }
}
}
}
if ($changed) { [void](Write-Log -Sessions @($sessions)) }
} catch {
Write-Err "handler_failed: $_" 'handler'
} finally {
if ($acquired -and $mutex) { try { $mutex.ReleaseMutex() | Out-Null } catch { } }
if ($mutex) { try { $mutex.Dispose() } catch { } }
}
exit 0
记录的日志内容大概为 d:/claude/logs/2026-06-23.json：
[
{
"project": "d:\\claude",
"session_id": "70a5a2a0-159d-4c17-8bad-84c402f93caf",
"prompts": [
{
"content": "你好",
"model": "ark-code-latest",
"start_time": "2026-06-23 17:05:50",
"end_time": "2026-06-23 17:05:53",
"duration": "00:00:03",
"status": "completed"
}
]
}
]【说明：本站主要是个人笔记和代码的分享，内容可能会不定期修改。目前文章支持直接转载，引用或转载请注明出处：https://www.guanjihuan.com 。本站采用知识共享署名许可协议 CC BY】
