feat(task-logger): strip ANSI escape codes from logs and extend coverage (#1411)

* auto-claude: subtask-1-1 - Add strip_ansi_codes() utility function to task_lo

* auto-claude: subtask-1-2 - Apply ANSI sanitization to TaskLogger.log_with_detail() and TaskLogger.tool_end()

Changes:
- Import strip_ansi_codes from utils in logger.py
- Apply strip_ansi_codes to detail parameter in log_with_detail()
- Apply strip_ansi_codes to stored_detail in tool_end()
- Fix circular import in utils.py using TYPE_CHECKING

Co-Authored-By: Claude <noreply@anthropic.com>

* auto-claude: subtask-1-3 - Apply ANSI sanitization to StreamingLogCapture.pro

- Import strip_ansi_codes from utils module
- Apply sanitization to process_text() method before logging
- Ensures ANSI escape codes are removed from streaming text output

Co-Authored-By: Claude <noreply@anthropic.com>

* feat(task-logger): strip ANSI escape codes from logs and extend coverage

Backend (Python):
- Extend CSI pattern to support private mode parameters (?<>=)
- Add None handling to strip_ansi_codes() function
- Add sanitization to TaskLogger.log() method (critical gap fix)
- Export strip_ansi_codes in task_logger public API
- Add 25 comprehensive unit tests for strip_ansi_codes()

Frontend (TypeScript):
- Extend CSI pattern to support private mode parameters (?<>=)
- Apply ANSI sanitization to merge preview error messages in Kanban

This ensures clean display of task logs and error messages in the UI
by removing terminal color/formatting escape sequences.

* fix(task_logger): resolve cyclic import issue

Move strip_ansi_codes import from module-level to local imports in
logger.py to avoid cyclic import when __init__.py imports both modules.
Also use lazy import in get_task_logger() to avoid cyclic import at module level.

Fixes NameError in test_planner_session_does_not_trigger_post_session_processing_on_retry

* fix(tests): add missing os import in test_task_logger.py

Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com>

* fix(task_logger): sanitize content parameter in log_with_detail()

- Apply ANSI stripping to both content and detail in log_with_detail()
- Remove unused Path import from test file
- Add test for content sanitization in log_with_detail()

Fixes issue identified by CodeRabbit review where log_with_detail()
only sanitized detail but not content, allowing ANSI codes to leak
into stored logs and UI components.

Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com>

* fix(task_logger): sanitize result and content in tool_end()

- Apply ANSI stripping to display_result and content in tool_end()
- Add test for result and content sanitization in tool_end()

Fixes CodeRabbit review comment where tool_end() only sanitized
detail but not result/content, allowing ANSI codes to leak into
the stored log content and UI components.

Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com>

* refactor: consolidate duplicate inline imports and remove redundant assert

- Consolidate 3 inline imports of strip_ansi_codes in tool_end() to single import
- Consolidate 2 inline imports in log_with_detail() to single import
- Remove redundant 'assert True' in test_public_api_exports

Addressed CodeRabbit review comments about code quality.

Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com>

* fix: scope strip_ansi_codes usage within import block

Move content and detail sanitization inside the import block to
resolve CodeQL false positive about potentially uninitialized
local variable. The import and all usages are now within
the same conditional scope.

Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com>

* fix: address CodeRabbit review comments

- Add lazy import of TaskLogger in update_task_logger_path() to avoid
  potential NameError when accessing TaskLogger.LOG_FILE
- Sanitize text before checking for empty in capture.process_text() to
  avoid logging blank entries when input contains only ANSI codes

These were false positives in practice but improve code clarity
and static analysis results.

Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com>

* fix: sanitize message, tool_input, and subphase parameters in logging methods

Apply strip_ansi_codes() to:
- start_phase() message parameter
- end_phase() message parameter
- tool_start() tool_input parameter
- start_subphase() subphase parameter

This ensures consistency with other logging methods (log, log_with_detail,
tool_end) which all sanitize input before storage.

Addresses Auto Claude PR Review findings:
- 🟡 [31465b234445] start_phase() message not sanitized
- 🟡 [18192bb9d19d] end_phase() message not sanitized
- 🟡 [16aa996a0d8d] tool_start() tool_input not sanitized
- 🟡 [b39df9833b80] start_subphase() subphase not sanitized

* refactor: extract ANSI utilities to separate module to fix cyclic import

Create new task_logger/ansi.py module containing strip_ansi_codes() and
related ANSI patterns, removing the dependency on logger.py.

This resolves CodeQL cyclic import warnings:
- utils.py imports TaskLogger (TYPE_CHECKING only)
- logger.py imports strip_ansi_codes from utils
- New ansi.py has no dependencies on other task_logger modules

Changes:
- Create apps/backend/task_logger/ansi.py with strip_ansi_codes()
- Update __init__.py to import strip_ansi_codes directly from .ansi
- Update logger.py to import from .ansi instead of .utils (7 locations)
- Update capture.py to import from .ansi instead of .utils
- Update tests to import from task_logger.ansi directly
- Remove duplicate ANSI code patterns from utils.py

All 27 tests pass.

* fix: address CodeRabbit review comments

1. Expand ANSI_CSI_PATTERN to match full ANSI/VT100 CSI final-byte range
   - Old pattern: \x1b\[[?><=0-9;]*[A-Za-z]
   - New pattern: \x1b\[[0-?]*[ -/]*[@-~]
   - Now strips bracketed paste sequences (\x1b[200~, \x1b[201~) and other
     CSI sequences with non-letter final bytes

2. Print sanitized phase_message instead of raw message in start_phase/end_phase
   - Prevents ANSI codes from appearing in console output
   - Keeps console output consistent with stored log content

3. Add test for bracketed paste sequences (test_csi_bracketed_paste)

All 28 tests pass.

* fix: address CodeRabbit review comments

1. Use platform abstraction in subprocess-spawn test
   - Replace process.platform === 'win32' with isWindows() from platform module
   - Follows coding guidelines for cross-platform checks

2. Add trailing newline to test_task_logger.py for POSIX compliance

All tests pass (28 Python tests, 14 TypeScript tests).

* fix: resolve all Auto Claude PR Review findings

Backend:
- Move strip_ansi_codes to module-level import in logger.py
- Removes 7 duplicate inline imports, keeping only 1 at top of file

Frontend:
- Update ANSI_CSI_PATTERN to match backend regex: /\x1b\[[0-?]*[ -/]*[@-~]/g
- Change final byte pattern from [A-Za-z] to [@-~] for full CSI coverage
- Add [ -/]* for intermediate bytes handling
- Add tests for bracketed paste sequences and private mode parameters

This resolves 4 remaining findings:
- 🔵 [LOW] Duplicate inline imports - FIXED
- 🟡 [MEDIUM] Frontend regex missing CSI final bytes - FIXED
- 🔵 [LOW] Frontend regex missing intermediate bytes - FIXED
- 🔵 [LOW] Frontend missing bracketed paste tests - FIXED

All tests pass: 28 Python + 36 TypeScript = 64 total.

* fix: sanitize before truncation to avoid partial ANSI remnants

Truncating before sanitizing can cut ANSI escape sequences in half,
leaving stray control characters that the regex won't remove.

Changes:
- Sanitize display_result before truncation (300 char limit)
- Sanitize stored_detail before truncation (10KB limit)
- Use original detail length for truncation message

Fixes CodeRabbit review comment about tool_end() sanitization order.

All 28 tests pass.

* refactor: address CodeRabbit nitpick comments

Backend:
- Remove redundant outer guard "if content or detail:" in log_with_detail()
- Remove redundant sanitization of content in tool_end() (already sanitized)
- Reduces nesting and removes unnecessary conditional checks

Frontend:
- Standardize all subprocess test timeouts to 30000ms for Windows CI
- Provides consistent margin for slower Windows CI environment

All tests pass: 28 Python + 14 TypeScript = 42 total.

* refactor: remove redundant conditions in start_phase and end_phase

Since phase_message always has a fallback default value, it can never be
falsy. The if phase_message: guards are unnecessary.

Simplifies code by:
- Removing redundant condition before sanitization
- Removing redundant condition before print

All 28 tests pass.

* fix: address Auto Claude review LOW severity findings

1. Fix truncation message to report sanitized length
   - Use sanitized_len for truncation message instead of unsanitized detail length
   - Ensures reported length matches visible content

2. Fix misleading comment in utils.py
   - Update comment to reflect that ANSI functions are in ansi.py module
   - Clarifies architectural decision rather than implying an import

All 28 tests pass.

* fix: apply ANSI sanitization to all subprocess error paths

Apply stripAnsiCodes() to merge and create PR error output
for consistency with the preview handler.

* refactor: remove unused variable original_len (dead code)

Remove the unused original_len variable that was leftover
from the previous ANSI sanitization fix.

---------

Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com>
Co-authored-by: StillKnotKnown <stillknotknown@users.noreply.github.com>
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Andy <119136210+AndyMik90@users.noreply.github.com>
This commit is contained in:
StillKnotKnown
2026-01-26 13:47:56 +02:00
committed by GitHub
parent 26c9083d3b
commit 988ec0c25b
11 changed files with 513 additions and 43 deletions
+8 -3
View File
@@ -14,6 +14,8 @@ Key features:
# Export models
# Export streaming capture
# Export utility functions
from .ansi import strip_ansi_codes
from .capture import StreamingLogCapture
# Export main logger
@@ -22,9 +24,11 @@ from .models import LogEntry, LogEntryType, LogPhase, PhaseLog
# Export storage utilities
from .storage import get_active_phase, load_task_logs
# Export utility functions
from .utils import clear_task_logger, get_task_logger, update_task_logger_path
from .utils import (
clear_task_logger,
get_task_logger,
update_task_logger_path,
)
__all__ = [
# Models
@@ -41,6 +45,7 @@ __all__ = [
"get_task_logger",
"clear_task_logger",
"update_task_logger_path",
"strip_ansi_codes",
# Streaming capture
"StreamingLogCapture",
]
+53
View File
@@ -0,0 +1,53 @@
"""
ANSI escape code utilities for task logging.
This module contains functions for stripping ANSI escape codes from strings.
It has no dependencies on other task_logger modules to avoid cyclic imports.
"""
import re
# ANSI escape code patterns
# ANSI CSI (Control Sequence Introducer) escape sequence pattern.
# Matches the full ANSI/VT100 CSI form: ESC [ parameter bytes (0-?) intermediate bytes ( -/) final bytes (@-~)
# Parameter bytes: 0x30-0x3F (digits 0-9, :;<=>?)
# Intermediate bytes: 0x20-0x2F (space and !"#$%&'()*+,-./)
# Final bytes: 0x40-0x7E (@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\]^_`abcdefghijklmnopqrstuvwxyz{|}~)
# Examples: \x1b[31m (red), \x1b[?25l (hide cursor), \x1b[200~ (bracketed paste start)
ANSI_CSI_PATTERN = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
# OSC (Operating System Command) escape sequences with BEL (bell) terminator
# Matches: \x1b] ... \x07
ANSI_OSC_BEL_PATTERN = re.compile(r"\x1b\][^\x07]*\x07")
# OSC (Operating System Command) escape sequences with ST (string terminator)
# Matches: \x1b] ... \x1b\
ANSI_OSC_ST_PATTERN = re.compile(r"\x1b\][^\x1b]*\x1b\\")
def strip_ansi_codes(text: str | None) -> str:
"""
Removes ANSI escape codes from a string.
These sequences are used for terminal coloring/formatting but appear
as raw text in logs and UI components.
Args:
text: The string potentially containing ANSI escape codes, or None
Returns:
The string with all ANSI escape sequences removed, or empty string if input is None
Example:
>>> strip_ansi_codes('\\x1b[90m[21:40:22.196]\\x1b[0m \\x1b[36m[DEBUG]\\x1b[0m')
'[21:40:22.196] [DEBUG]'
"""
if not text:
return ""
# Remove all ANSI escape sequences
result = ANSI_CSI_PATTERN.sub("", text)
result = ANSI_OSC_BEL_PATTERN.sub("", result)
result = ANSI_OSC_ST_PATTERN.sub("", result)
return result
+5 -2
View File
@@ -2,6 +2,7 @@
Streaming log capture for agent sessions.
"""
from .ansi import strip_ansi_codes
from .logger import TaskLogger
from .models import LogPhase
@@ -36,8 +37,10 @@ class StreamingLogCapture:
def process_text(self, text: str) -> None:
"""Process text output from the agent."""
if text.strip():
self.logger.log(text, phase=self.phase)
# Remove ANSI escape codes before logging
sanitized_text = strip_ansi_codes(text)
if sanitized_text.strip():
self.logger.log(sanitized_text, phase=self.phase)
def process_tool_start(self, tool_name: str, tool_input: str | None = None) -> None:
"""Process tool start."""
+33 -10
View File
@@ -7,6 +7,7 @@ from pathlib import Path
from core.debug import debug, debug_error, debug_info, debug_success, is_debug_enabled
from .ansi import strip_ansi_codes
from .models import LogEntry, LogEntryType, LogPhase
from .storage import LogStorage
from .streaming import emit_marker
@@ -160,6 +161,7 @@ class TaskLogger:
# Add phase start entry
phase_message = message or f"Starting {phase_key} phase"
phase_message = strip_ansi_codes(phase_message)
entry = LogEntry(
timestamp=self._timestamp(),
type=LogEntryType.PHASE_START.value,
@@ -172,9 +174,8 @@ class TaskLogger:
# Debug log (when DEBUG=true)
self._debug_log(phase_message, LogEntryType.PHASE_START, phase_key)
# Also print the message
if message:
print(message, flush=True)
# Also print the message (sanitized)
print(phase_message, flush=True)
def end_phase(
self, phase: LogPhase, success: bool = True, message: str | None = None
@@ -203,6 +204,8 @@ class TaskLogger:
phase_message = (
message or f"{'Completed' if success else 'Failed'} {phase_key} phase"
)
phase_message = strip_ansi_codes(phase_message)
entry = LogEntry(
timestamp=self._timestamp(),
type=LogEntryType.PHASE_END.value,
@@ -216,8 +219,8 @@ class TaskLogger:
entry_type = LogEntryType.SUCCESS if success else LogEntryType.ERROR
self._debug_log(phase_message, entry_type, phase_key)
if message:
print(message, flush=True)
# Print the message (sanitized)
print(phase_message, flush=True)
if phase == self.current_phase:
self.current_phase = None
@@ -240,6 +243,10 @@ class TaskLogger:
phase: Optional phase override (uses current_phase if not specified)
print_to_console: Whether to also print to stdout (default True)
"""
# Sanitize content to remove ANSI escape codes before storage
if content:
content = strip_ansi_codes(content)
phase_key = (phase or self.current_phase or LogPhase.CODING).value
entry = LogEntry(
@@ -307,6 +314,13 @@ class TaskLogger:
"""
phase_key = (phase or self.current_phase or LogPhase.CODING).value
# Sanitize content and detail before storage
if content:
content = strip_ansi_codes(content)
if detail:
detail = strip_ansi_codes(detail)
entry = LogEntry(
timestamp=self._timestamp(),
type=entry_type.value,
@@ -363,6 +377,10 @@ class TaskLogger:
"""
phase_key = (phase or self.current_phase or LogPhase.CODING).value
# Sanitize subphase before use
if subphase:
subphase = strip_ansi_codes(subphase)
entry = LogEntry(
timestamp=self._timestamp(),
type=LogEntryType.INFO.value,
@@ -406,6 +424,10 @@ class TaskLogger:
"""
phase_key = (phase or self.current_phase or LogPhase.CODING).value
# Sanitize tool_input before use
if tool_input:
tool_input = strip_ansi_codes(tool_input)
# Truncate long inputs for display (increased limit to avoid hiding critical info)
display_input = tool_input
if display_input and len(display_input) > 300:
@@ -462,8 +484,8 @@ class TaskLogger:
"""
phase_key = (phase or self.current_phase or LogPhase.CODING).value
# Truncate long results for display (increased limit to avoid hiding critical info)
display_result = result
# Sanitize before truncation to avoid cutting ANSI sequences mid-stream
display_result = strip_ansi_codes(result) if result else None
if display_result and len(display_result) > 300:
display_result = display_result[:297] + "..."
@@ -472,12 +494,13 @@ class TaskLogger:
if display_result:
content += f": {display_result}"
# Truncate detail for storage (max 10KB to avoid bloating JSON)
stored_detail = detail
# Sanitize before truncating detail
stored_detail = strip_ansi_codes(detail) if detail else None
if stored_detail and len(stored_detail) > 10240:
sanitized_len = len(stored_detail)
stored_detail = (
stored_detail[:10240]
+ f"\n\n... [truncated - full output was {len(detail)} chars]"
+ f"\n\n... [truncated - full output was {sanitized_len} chars]"
)
entry = LogEntry(
+14 -3
View File
@@ -3,16 +3,21 @@ Utility functions for task logging.
"""
from pathlib import Path
from typing import TYPE_CHECKING
# ANSI functions are in separate ansi.py module to avoid cyclic imports
if TYPE_CHECKING:
from .logger import TaskLogger
from .logger import TaskLogger
# Global logger instance for easy access
_current_logger: TaskLogger | None = None
_current_logger: "TaskLogger | None" = None
def get_task_logger(
spec_dir: Path | None = None, emit_markers: bool = True
) -> TaskLogger | None:
) -> "TaskLogger | None":
"""
Get or create a task logger for the given spec directory.
@@ -29,6 +34,9 @@ def get_task_logger(
return _current_logger
if _current_logger is None or _current_logger.spec_dir != spec_dir:
# Lazy import to avoid cyclic import
from .logger import TaskLogger
_current_logger = TaskLogger(spec_dir, emit_markers)
return _current_logger
@@ -55,6 +63,9 @@ def update_task_logger_path(new_spec_dir: Path) -> None:
if _current_logger is None:
return
# Lazy import to avoid cyclic import
from .logger import TaskLogger
# Update the logger's internal paths
_current_logger.spec_dir = Path(new_spec_dir)
_current_logger.log_file = _current_logger.spec_dir / TaskLogger.LOG_FILE
@@ -12,6 +12,7 @@ import { mkdirSync, rmSync, existsSync, writeFileSync, mkdtempSync } from 'fs';
import { tmpdir } from 'os';
import path from 'path';
import { findPythonCommand, parsePythonCommand } from '../../main/python-detector';
import { isWindows } from '../../main/platform';
// Test directories - use secure temp directory with random suffix
let TEST_DIR: string;
@@ -361,7 +362,7 @@ describe('Subprocess Spawn Integration', () => {
expect(result).toBe(true);
// On Windows, kill() is called without arguments; on Unix, kill('SIGTERM') is used
if (process.platform === 'win32') {
if (isWindows()) {
expect(mockProcess.kill).toHaveBeenCalled();
} else {
expect(mockProcess.kill).toHaveBeenCalledWith('SIGTERM');
@@ -21,6 +21,7 @@ import {
import { persistPlanStatus, updateTaskMetadataPrUrl } from './plan-file-utils';
import { getIsolatedGitEnv, detectWorktreeBranch, refreshGitIndex } from '../../utils/git-isolation';
import { killProcessGracefully } from '../../platform';
import { stripAnsiCodes } from '../../../shared/utils/ansi-sanitizer';
// Regex pattern for validating git branch names
const GIT_BRANCH_REGEX = /^[a-zA-Z0-9][a-zA-Z0-9._/-]*[a-zA-Z0-9]$|^[a-zA-Z0-9]$/;
@@ -2328,7 +2329,9 @@ export function registerWorktreeHandlers(
success: true,
data: {
success: false,
message: hasConflicts ? 'Merge conflicts detected' : `Merge failed: ${stderr || stdout}`,
message: hasConflicts
? 'Merge conflicts detected'
: `Merge failed: ${stripAnsiCodes(stderr || stdout)}`,
conflictFiles: hasConflicts ? [] : undefined
}
});
@@ -2528,7 +2531,7 @@ export function registerWorktreeHandlers(
console.error('[IPC] stderr:', stderr);
resolve({
success: false,
error: `Failed to parse preview result: ${stderr || stdout}`
error: `Failed to parse preview result: ${stripAnsiCodes(stderr || stdout)}`
});
}
} else {
@@ -2537,7 +2540,7 @@ export function registerWorktreeHandlers(
console.error('[IPC] stdout:', stdout);
resolve({
success: false,
error: `Preview failed: ${stderr || stdout}`
error: `Preview failed: ${stripAnsiCodes(stderr || stdout)}`
});
}
});
@@ -3129,7 +3132,7 @@ export function registerWorktreeHandlers(
// Prefer stdout over stderr since stderr often contains debug messages
resolve({
success: false,
error: stdout || stderr || 'Failed to create PR'
error: stripAnsiCodes(stdout || stderr || 'Failed to create PR')
});
}
}
@@ -44,6 +44,20 @@ describe('stripAnsiCodes', () => {
const expected = 'Text';
expect(stripAnsiCodes(input)).toBe(expected);
});
it('should handle CSI bracketed paste sequences', () => {
// Bracketed paste start/end with non-alphabetic final byte (~)
const input = '\x1b[200~pasted text\x1b[201~';
const expected = 'pasted text';
expect(stripAnsiCodes(input)).toBe(expected);
});
it('should handle CSI with private mode parameters', () => {
// Private mode sequences use ?<>=
const input = '\x1b[?25lhide cursor\x1b[?25hshow cursor';
const expected = 'hide cursorshow cursor';
expect(stripAnsiCodes(input)).toBe(expected);
});
});
describe('OSC (Operating System Command) patterns', () => {
@@ -12,14 +12,15 @@
/**
* ANSI CSI (Control Sequence Introducer) escape sequence pattern.
* Matches:
* - \x1b[ or \e[ or \033[ - Escape sequence start
* - [0-9;]* - Parameter bytes (numbers and semicolons)
* - [A-Za-z] - Final byte (command)
* - m - Specifically for SGR (Select Graphic Rendition) color codes
* Matches the full ANSI/VT100 CSI form: ESC [ parameter-bytes intermediate-bytes final-bytes
* - Parameter bytes: 0x30-0x3F (digits 0-9, :;<=>?) -> [0-?]* in regex
* - Intermediate bytes: 0x20-0x2F (space and !"#$%&'()*+,-./) -> [ -/]* in regex
* - Final bytes: 0x40-0x7E (@ through ~) -> [@-~] in regex
*
* Examples: \x1b[31m (red), \x1b[?25l (hide cursor), \x1b[200~ (bracketed paste start)
*/
// biome-ignore lint/suspicious/noControlCharactersInRegex: ANSI escape codes require control characters
const ANSI_CSI_PATTERN = /\x1b\[[0-9;]*[A-Za-z]/g;
const ANSI_CSI_PATTERN = /\x1b\[[0-?]*[ -/]*[@-~]/g;
/**
* OSC (Operating System Command) escape sequences.
+32 -14
View File
@@ -247,6 +247,7 @@
"integrity": "sha512-H3mcG6ZDLTlYfaSNi0iOKkigqMFvkTKlGUYlD8GW7nNOYRrevuA46iTypPyv+06V3fEmvvazfntkBU34L0azAw==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"@babel/code-frame": "^7.28.6",
"@babel/generator": "^7.28.6",
@@ -811,6 +812,7 @@
}
],
"license": "MIT",
"peer": true,
"engines": {
"node": ">=18"
},
@@ -854,6 +856,7 @@
}
],
"license": "MIT",
"peer": true,
"engines": {
"node": ">=18"
}
@@ -893,6 +896,7 @@
"resolved": "https://registry.npmjs.org/@dnd-kit/core/-/core-6.3.1.tgz",
"integrity": "sha512-xkGBRQQab4RLwgXxoqETICr6S5JlogafbhNsidmrkVv2YRs5MLwpjoF2qpiGjQt8S9AoxtIV603s0GIUpY5eYQ==",
"license": "MIT",
"peer": true,
"dependencies": {
"@dnd-kit/accessibility": "^3.1.1",
"@dnd-kit/utilities": "^3.2.2",
@@ -1311,7 +1315,6 @@
"dev": true,
"license": "BSD-2-Clause",
"optional": true,
"peer": true,
"dependencies": {
"cross-dirname": "^0.1.0",
"debug": "^4.3.4",
@@ -1333,7 +1336,6 @@
"dev": true,
"license": "MIT",
"optional": true,
"peer": true,
"dependencies": {
"graceful-fs": "^4.2.0",
"jsonfile": "^6.0.1",
@@ -1350,7 +1352,6 @@
"dev": true,
"license": "MIT",
"optional": true,
"peer": true,
"dependencies": {
"universalify": "^2.0.0"
},
@@ -1365,7 +1366,6 @@
"dev": true,
"license": "MIT",
"optional": true,
"peer": true,
"engines": {
"node": ">= 10.0.0"
}
@@ -2245,6 +2245,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.9.0.tgz",
"integrity": "sha512-3giAOQvZiH5F9bMlMiv8+GSPMeqg0dbaeo58/0SlA9sxSqZhnUtxzX9/2FzyhS9sWQf5S0GJE0AKBrFqjpeYcg==",
"license": "Apache-2.0",
"peer": true,
"engines": {
"node": ">=8.0.0"
}
@@ -2266,6 +2267,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/context-async-hooks/-/context-async-hooks-2.4.0.tgz",
"integrity": "sha512-jn0phJ+hU7ZuvaoZE/8/Euw3gvHJrn2yi+kXrymwObEPVPjtwCmkvXDRQCWli+fCTTF/aSOtXaLr7CLIvv3LQg==",
"license": "Apache-2.0",
"peer": true,
"engines": {
"node": "^18.19.0 || >=20.6.0"
},
@@ -2278,6 +2280,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/core/-/core-2.4.0.tgz",
"integrity": "sha512-KtcyFHssTn5ZgDu6SXmUznS80OFs/wN7y6MyFRRcKU6TOw8hNcGxKvt8hsdaLJfhzUszNSjURetq5Qpkad14Gw==",
"license": "Apache-2.0",
"peer": true,
"dependencies": {
"@opentelemetry/semantic-conventions": "^1.29.0"
},
@@ -2293,6 +2296,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/instrumentation/-/instrumentation-0.208.0.tgz",
"integrity": "sha512-Eju0L4qWcQS+oXxi6pgh7zvE2byogAkcsVv0OjHF/97iOz1N/aKE6etSGowYkie+YA1uo6DNwdSxaaNnLvcRlA==",
"license": "Apache-2.0",
"peer": true,
"dependencies": {
"@opentelemetry/api-logs": "0.208.0",
"import-in-the-middle": "^2.0.0",
@@ -2695,6 +2699,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/resources/-/resources-2.4.0.tgz",
"integrity": "sha512-RWvGLj2lMDZd7M/5tjkI/2VHMpXebLgPKvBUd9LRasEWR2xAynDwEYZuLvY9P2NGG73HF07jbbgWX2C9oavcQg==",
"license": "Apache-2.0",
"peer": true,
"dependencies": {
"@opentelemetry/core": "2.4.0",
"@opentelemetry/semantic-conventions": "^1.29.0"
@@ -2711,6 +2716,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/sdk-trace-base/-/sdk-trace-base-2.4.0.tgz",
"integrity": "sha512-WH0xXkz/OHORDLKqaxcUZS0X+t1s7gGlumr2ebiEgNZQl2b0upK2cdoD0tatf7l8iP74woGJ/Kmxe82jdvcWRw==",
"license": "Apache-2.0",
"peer": true,
"dependencies": {
"@opentelemetry/core": "2.4.0",
"@opentelemetry/resources": "2.4.0",
@@ -2728,6 +2734,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/semantic-conventions/-/semantic-conventions-1.39.0.tgz",
"integrity": "sha512-R5R9tb2AXs2IRLNKLBJDynhkfmx7mX0vi8NkhZb3gUkPWHn6HXk5J8iQ/dql0U3ApfWym4kXXmBDRGO+oeOfjg==",
"license": "Apache-2.0",
"peer": true,
"engines": {
"node": ">=14"
}
@@ -4922,6 +4929,7 @@
"integrity": "sha512-o4PXJQidqJl82ckFaXUeoAW+XysPLauYI43Abki5hABd853iMhitooc6znOnczgbTYmEP6U6/y1ZyKAIsvMKGg==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"@babel/code-frame": "^7.10.4",
"@babel/runtime": "^7.12.5",
@@ -5220,6 +5228,7 @@
"resolved": "https://registry.npmjs.org/@types/react/-/react-19.2.8.tgz",
"integrity": "sha512-3MbSL37jEchWZz2p2mjntRZtPt837ij10ApxKfgmXCTuHWagYg7iA5bqPw6C8BMPfwidlvfPI/fxOc42HLhcyg==",
"license": "MIT",
"peer": true,
"dependencies": {
"csstype": "^3.2.2"
}
@@ -5230,6 +5239,7 @@
"integrity": "sha512-jp2L/eY6fn+KgVVQAOqYItbF0VY/YApe5Mz2F0aykSO8gx31bYCZyvSeYxCHKvzHG5eZjc+zyaS5BrBWya2+kQ==",
"devOptional": true,
"license": "MIT",
"peer": true,
"peerDependencies": {
"@types/react": "^19.2.0"
}
@@ -5495,6 +5505,7 @@
"resolved": "https://registry.npmjs.org/acorn/-/acorn-8.15.0.tgz",
"integrity": "sha512-NZyJarBfL7nWwIq+FDL6Zp/yHEhePMNnnJ0y3qfieCrmNvYct8uvtiV41UvlSe6apAfk0fY1FbWx+NwfmpvtTg==",
"license": "MIT",
"peer": true,
"bin": {
"acorn": "bin/acorn"
},
@@ -5527,6 +5538,7 @@
"integrity": "sha512-j3fVLgvTo527anyYyJOGTYJbG+vnnQYvE0m5mmkc1TK+nxAppkCLMIL0aZ4dblVCNoGShhm+kzE4ZUykBoMg4g==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"fast-deep-equal": "^3.1.1",
"fast-json-stable-stringify": "^2.0.0",
@@ -5968,6 +5980,7 @@
}
],
"license": "MIT",
"peer": true,
"dependencies": {
"baseline-browser-mapping": "^2.9.0",
"caniuse-lite": "^1.0.30001759",
@@ -6666,8 +6679,7 @@
"integrity": "sha512-+R08/oI0nl3vfPcqftZRpytksBXDzOUveBq/NBVx0sUp1axwzPQrKinNx5yd5sxPu8j1wIy8AfnVQ+5eFdha6Q==",
"dev": true,
"license": "MIT",
"optional": true,
"peer": true
"optional": true
},
"node_modules/cross-env": {
"version": "10.1.0",
@@ -6984,6 +6996,7 @@
"integrity": "sha512-ce4Ogns4VMeisIuCSK0C62umG0lFy012jd8LMZ6w/veHUeX4fqfDrGe+HTWALAEwK6JwKP+dhPvizhArSOsFbg==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"app-builder-lib": "26.4.0",
"builder-util": "26.3.4",
@@ -7140,6 +7153,7 @@
"dev": true,
"hasInstallScript": true,
"license": "MIT",
"peer": true,
"dependencies": {
"@electron/get": "^2.0.0",
"@types/node": "^22.7.7",
@@ -7388,7 +7402,6 @@
"dev": true,
"hasInstallScript": true,
"license": "MIT",
"peer": true,
"dependencies": {
"@electron/asar": "^3.2.1",
"debug": "^4.1.1",
@@ -7409,7 +7422,6 @@
"integrity": "sha512-YJDaCJZEnBmcbw13fvdAM9AwNOJwOzrE4pqMqBq5nFiEqXUqHwlK4B+3pUw6JNvfSPtX05xFHtYy/1ni01eGCw==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"graceful-fs": "^4.1.2",
"jsonfile": "^4.0.0",
@@ -8419,6 +8431,7 @@
}
],
"license": "MIT",
"peer": true,
"dependencies": {
"@babel/runtime": "^7.28.4"
},
@@ -8745,6 +8758,7 @@
"integrity": "sha512-mjzqwWRD9Y1J1KUi7W97Gja1bwOOM5Ug0EZ6UDK3xS7j7mndrkwozHtSblfomlzyB4NepioNt+B2sOSzczVgtQ==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"@acemir/cssom": "^0.9.28",
"@asamuzakjp/dom-selector": "^6.7.6",
@@ -11260,6 +11274,7 @@
}
],
"license": "MIT",
"peer": true,
"dependencies": {
"nanoid": "^3.3.11",
"picocolors": "^1.1.1",
@@ -11335,7 +11350,6 @@
"dev": true,
"license": "MIT",
"optional": true,
"peer": true,
"dependencies": {
"commander": "^9.4.0"
},
@@ -11353,7 +11367,6 @@
"dev": true,
"license": "MIT",
"optional": true,
"peer": true,
"engines": {
"node": "^12.20.0 || >=14"
}
@@ -11480,6 +11493,7 @@
"resolved": "https://registry.npmjs.org/react/-/react-19.2.3.tgz",
"integrity": "sha512-Ku/hhYbVjOQnXDZFv2+RibmLFGwFdeeKHFcOTlrt7xplBnya5OGn/hIRDsqDiSUcfORsDC7MPxwork8jBwsIWA==",
"license": "MIT",
"peer": true,
"engines": {
"node": ">=0.10.0"
}
@@ -11489,6 +11503,7 @@
"resolved": "https://registry.npmjs.org/react-dom/-/react-dom-19.2.3.tgz",
"integrity": "sha512-yELu4WmLPw5Mr/lmeEpox5rw3RETacE++JgHqQzd2dg+YbJuat3jH4ingc+WPZhxaoFzdv9y33G+F7Nl5O0GBg==",
"license": "MIT",
"peer": true,
"dependencies": {
"scheduler": "^0.27.0"
},
@@ -11891,7 +11906,6 @@
"deprecated": "Rimraf versions prior to v4 are no longer supported",
"dev": true,
"license": "ISC",
"peer": true,
"dependencies": {
"glob": "^7.1.3"
},
@@ -12470,7 +12484,8 @@
"version": "4.1.18",
"resolved": "https://registry.npmjs.org/tailwindcss/-/tailwindcss-4.1.18.tgz",
"integrity": "sha512-4+Z+0yiYyEtUVCScyfHCxOYP06L5Ne+JiHhY2IjR2KWMIWhJOYZKLSGZaP5HkZ8+bY0cxfzwDE5uOmzFXyIwxw==",
"license": "MIT"
"license": "MIT",
"peer": true
},
"node_modules/tapable": {
"version": "2.3.0",
@@ -12573,7 +12588,6 @@
"integrity": "sha512-yYrrsWnrXMcdsnu/7YMYAofM1ktpL5By7vZhf15CrXijWWrEYZks5AXBudalfSWJLlnen/QUJUB5aoB0kqZUGA==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"mkdirp": "^0.5.1",
"rimraf": "~2.6.2"
@@ -12637,7 +12651,6 @@
"integrity": "sha512-FP+p8RB8OWpF3YZBCrP5gtADmtXApB5AMLn+vdyA+PyxCjrCs00mjyUozssO33cwDeT3wNGdLxJ5M//YqtHAJw==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"minimist": "^1.2.6"
},
@@ -12729,6 +12742,7 @@
"integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
"dev": true,
"license": "MIT",
"peer": true,
"engines": {
"node": ">=12"
},
@@ -12887,6 +12901,7 @@
"integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==",
"devOptional": true,
"license": "Apache-2.0",
"peer": true,
"bin": {
"tsc": "bin/tsc",
"tsserver": "bin/tsserver"
@@ -13193,6 +13208,7 @@
"integrity": "sha512-w+N7Hifpc3gRjZ63vYBXA56dvvRlNWRczTdmCBBa+CotUzAPf5b7YMdMR/8CQoeYE5LX3W4wj6RYTgonm1b9DA==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"esbuild": "^0.27.0",
"fdir": "^6.5.0",
@@ -13785,6 +13801,7 @@
"integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
"dev": true,
"license": "MIT",
"peer": true,
"engines": {
"node": ">=12"
},
@@ -14226,6 +14243,7 @@
"resolved": "https://registry.npmjs.org/zod/-/zod-4.3.5.tgz",
"integrity": "sha512-k7Nwx6vuWx1IJ9Bjuf4Zt1PEllcwe7cls3VNzm4CQ1/hgtFUK2bRNG3rvnpPUhFjmqJKAKtjV576KnUkHocg/g==",
"license": "MIT",
"peer": true,
"funding": {
"url": "https://github.com/sponsors/colinhacks"
}
+338
View File
@@ -0,0 +1,338 @@
"""
Task Logger Tests
Tests for the task_logger module including ANSI code stripping functionality.
"""
import json
import os
import sys
# Add backend to path for imports
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'apps', 'backend'))
from task_logger.ansi import strip_ansi_codes
from task_logger.capture import StreamingLogCapture
from task_logger.logger import TaskLogger
from task_logger.models import LogEntryType, LogPhase
# ============================================================================
# Unit Tests for strip_ansi_codes() Function
# ============================================================================
class TestStripAnsiCodes:
"""Unit tests for the strip_ansi_codes() utility function."""
def test_empty_string(self):
"""Empty string should return empty string."""
assert strip_ansi_codes("") == ""
def test_none_input(self):
"""None input should return empty string."""
assert strip_ansi_codes(None) == ""
def test_no_ansi_codes(self):
"""Plain text without ANSI codes should be unchanged."""
assert strip_ansi_codes("plain text") == "plain text"
assert strip_ansi_codes("Hello, World!") == "Hello, World!"
assert strip_ansi_codes("12345") == "12345"
def test_simple_color_code(self):
"""Simple CSI color codes should be removed."""
assert strip_ansi_codes("\x1b[31mred\x1b[0m") == "red"
assert strip_ansi_codes("\x1b[32mgreen\x1b[0m") == "green"
assert strip_ansi_codes("\x1b[34mblue\x1b[0m") == "blue"
def test_vitest_like_output(self):
"""Vitest-like timestamp and debug output should be cleaned."""
input_text = "\x1b[90m[21:40:22.196]\x1b[0m \x1b[36m[DEBUG]\x1b[0m Test message"
expected = "[21:40:22.196] [DEBUG] Test message"
assert strip_ansi_codes(input_text) == expected
def test_multiple_ansi_codes(self):
"""Multiple consecutive ANSI codes should all be removed."""
input_text = "\x1b[31m\x1b[1mbold red\x1b[0m"
expected = "bold red"
assert strip_ansi_codes(input_text) == expected
def test_osc_bel_sequence(self):
"""OSC sequences with BEL terminator should be removed."""
assert strip_ansi_codes("\x1b]0;Window Title\x07") == ""
assert strip_ansi_codes("Text\x1b]0;Title\x07More") == "TextMore"
def test_osc_st_sequence(self):
"""OSC sequences with ST terminator should be removed."""
assert strip_ansi_codes("\x1b]0;Window Title\x1b\\") == ""
assert strip_ansi_codes("Text\x1b]0;Title\x1b\\More") == "TextMore"
def test_mixed_ansi_types(self):
"""Mixed CSI and OSC sequences in same string should all be removed."""
input_text = "\x1b[31mError:\x1b[0m \x1b]1;Title\x07Failed"
expected = "Error: Failed"
assert strip_ansi_codes(input_text) == expected
def test_multiline_text(self):
"""Multi-line text with ANSI codes should be cleaned."""
input_text = "\x1b[31mLine 1\x1b[0m\nLine 2\x1b[32m\x1b[1m\x1b[0m\nLine 3"
expected = "Line 1\nLine 2\nLine 3"
assert strip_ansi_codes(input_text) == expected
def test_private_mode_parameters(self):
"""CSI sequences with private mode parameters should be removed."""
# Cursor hide/show
assert strip_ansi_codes("\x1b[?25lHide\x1b[?25hShow") == "HideShow"
# Private mode with other chars
assert strip_ansi_codes("\x1b[=1hApplication Mode\x1b[=0l") == "Application Mode"
def test_csi_with_parameters(self):
"""CSI sequences with semicolon-separated parameters should be removed."""
# Bold red (1;31)
assert strip_ansi_codes("\x1b[1;31mText\x1b[0m") == "Text"
# Multiple parameters
assert strip_ansi_codes("\x1b[38;2;255;0;0mRGB Red\x1b[0m") == "RGB Red"
def test_csi_cursor_movement(self):
"""CSI cursor movement sequences should be removed."""
assert strip_ansi_codes("Text\x1b[2K") == "Text"
assert strip_ansi_codes("\x1b[0G\x1b[2KClear line") == "Clear line"
assert strip_ansi_codes("\x1b[A\x1b[B\x1b[C\x1b[D") == ""
def test_ansi_hyperlinks(self):
"""ANSI hyperlink format (OSC 8) should be removed."""
input_text = "\x1b]8;;https://example.com\x07Click here\x1b]8;;\x07"
expected = "Click here"
assert strip_ansi_codes(input_text) == expected
def test_csi_bracketed_paste(self):
"""CSI bracketed paste sequences should be removed (final byte ~)."""
# Bracketed paste start/end
assert strip_ansi_codes("\x1b[200~") == ""
assert strip_ansi_codes("\x1b[201~") == ""
# Bracketed paste with content
assert strip_ansi_codes("\x1b[200~text\x1b[201~") == "text"
def test_unicode_with_ansi(self):
"""Unicode text combined with ANSI codes should preserve Unicode."""
input_text = "\x1b[31m你好\x1b[0m \x1b[32m世界\x1b[0m"
expected = "你好 世界"
assert strip_ansi_codes(input_text) == expected
# Emoji
input_text = "\x1b[36m🎉\x1b[0m \x1b[33m🚀\x1b[0m"
expected = "🎉 🚀"
assert strip_ansi_codes(input_text) == expected
def test_very_long_input(self):
"""Very long strings with many ANSI codes should be handled efficiently."""
# Create a long string with alternating ANSI codes and text
parts = []
for i in range(100):
parts.append(f"\x1b[{i % 10}mtext{i}\x1b[0m")
input_text = "".join(parts)
result = strip_ansi_codes(input_text)
# Verify all ANSI codes are removed
assert "\x1b" not in result
# Verify text content is preserved
for i in range(100):
assert f"text{i}" in result
def test_only_ansi_codes(self):
"""String consisting entirely of ANSI codes should return empty."""
assert strip_ansi_codes("\x1b[31m\x1b[1m\x1b[4m") == ""
assert strip_ansi_codes("\x1b]0;Title\x07") == ""
def test_nested_ansi_sequences(self):
"""Nested ANSI sequences should all be removed."""
input_text = "\x1b[31m\x1b[1mbold red\x1b[0m \x1b[32mgreen\x1b[0m"
expected = "bold red green"
assert strip_ansi_codes(input_text) == expected
# ============================================================================
# Integration Tests for TaskLogger
# ============================================================================
class TestTaskLoggerAnsiIntegration:
"""Integration tests for TaskLogger ANSI code sanitization."""
def test_log_sanitizes_content(self, tmp_path):
"""The log() method should sanitize content before storage."""
logger = TaskLogger(tmp_path, emit_markers=False)
logger.log(
"\x1b[31mError message\x1b[0m",
LogEntryType.ERROR,
print_to_console=False
)
# Load the log file and verify content is sanitized
log_file = tmp_path / "task_logs.json"
with open(log_file) as f:
logs = json.load(f)
coding_entries = logs["phases"]["coding"]["entries"]
assert len(coding_entries) == 1
assert coding_entries[0]["content"] == "Error message"
assert "\x1b" not in coding_entries[0]["content"]
def test_log_with_detail_sanitizes_detail(self, tmp_path):
"""log_with_detail() should sanitize detail parameter."""
logger = TaskLogger(tmp_path, emit_markers=False)
logger.log_with_detail(
content="Reading file",
detail="\x1b[31mERROR:\x1b[0m File not found",
print_to_console=False
)
log_file = tmp_path / "task_logs.json"
with open(log_file) as f:
logs = json.load(f)
coding_entries = logs["phases"]["coding"]["entries"]
assert len(coding_entries) == 1
assert coding_entries[0]["detail"] == "ERROR: File not found"
assert "\x1b" not in coding_entries[0]["detail"]
def test_log_with_detail_sanitizes_content(self, tmp_path):
"""log_with_detail() should sanitize content parameter."""
logger = TaskLogger(tmp_path, emit_markers=False)
logger.log_with_detail(
content="\x1b[33mWarning:\x1b[0m Check this",
detail="Some detail text",
print_to_console=False
)
log_file = tmp_path / "task_logs.json"
with open(log_file) as f:
logs = json.load(f)
coding_entries = logs["phases"]["coding"]["entries"]
assert len(coding_entries) == 1
assert coding_entries[0]["content"] == "Warning: Check this"
assert "\x1b" not in coding_entries[0]["content"]
def test_tool_end_sanitizes_detail(self, tmp_path):
"""tool_end() should sanitize detail parameter."""
logger = TaskLogger(tmp_path, emit_markers=False)
logger.tool_start("Bash", "npm test")
logger.tool_end(
"Bash",
success=True,
result="Tests completed",
detail="\x1b[36m$ npm test\x1b[0m\n\x1b[32mPASS\x1b[0m All tests passed"
)
log_file = tmp_path / "task_logs.json"
with open(log_file) as f:
logs = json.load(f)
coding_entries = logs["phases"]["coding"]["entries"]
# Find the tool_end entry
tool_end_entries = [e for e in coding_entries if e["type"] == "tool_end"]
assert len(tool_end_entries) == 1
assert tool_end_entries[0]["detail"] == "$ npm test\nPASS All tests passed"
assert "\x1b" not in tool_end_entries[0]["detail"]
def test_tool_end_sanitizes_result_and_content(self, tmp_path):
"""tool_end() should sanitize result and content parameters."""
logger = TaskLogger(tmp_path, emit_markers=False)
logger.tool_start("Bash", "npm test")
logger.tool_end(
"Bash",
success=True,
result="\x1b[32mTests passed\x1b[0m",
detail="Some output"
)
log_file = tmp_path / "task_logs.json"
with open(log_file) as f:
logs = json.load(f)
coding_entries = logs["phases"]["coding"]["entries"]
tool_end_entries = [e for e in coding_entries if e["type"] == "tool_end"]
assert len(tool_end_entries) == 1
# Content should be "[Bash] Done: Tests passed" without ANSI codes
assert tool_end_entries[0]["content"] == "[Bash] Done: Tests passed"
assert "\x1b" not in tool_end_entries[0]["content"]
# ============================================================================
# Integration Tests for StreamingLogCapture
# ============================================================================
class TestStreamingLogCaptureAnsiIntegration:
"""Integration tests for StreamingLogCapture ANSI code sanitization."""
def test_process_text_sanitizes(self, tmp_path):
"""process_text() should sanitize text before logging."""
logger = TaskLogger(tmp_path, emit_markers=False)
with StreamingLogCapture(logger, LogPhase.CODING) as capture:
capture.process_text("\x1b[90m[DEBUG]\x1b[0m Processing...")
log_file = tmp_path / "task_logs.json"
with open(log_file) as f:
logs = json.load(f)
coding_entries = logs["phases"]["coding"]["entries"]
assert len(coding_entries) == 1
assert coding_entries[0]["content"] == "[DEBUG] Processing..."
assert "\x1b" not in coding_entries[0]["content"]
def test_process_text_multiple_calls(self, tmp_path):
"""Multiple process_text calls should each sanitize."""
logger = TaskLogger(tmp_path, emit_markers=False)
with StreamingLogCapture(logger, LogPhase.CODING) as capture:
capture.process_text("\x1b[31mError\x1b[0m")
capture.process_text("\x1b[32mSuccess\x1b[0m")
log_file = tmp_path / "task_logs.json"
with open(log_file) as f:
logs = json.load(f)
coding_entries = logs["phases"]["coding"]["entries"]
assert len(coding_entries) == 2
assert coding_entries[0]["content"] == "Error"
assert coding_entries[1]["content"] == "Success"
# ============================================================================
# Public API Tests
# ============================================================================
class TestTaskLoggerPublicAPI:
"""Tests for the task_logger public API exports."""
def test_strip_ansi_codes_is_exported(self):
"""strip_ansi_codes should be importable from task_logger package."""
from task_logger import strip_ansi_codes as exported_strip
# Verify it's the same function
assert exported_strip is strip_ansi_codes
# Verify it works
assert exported_strip("\x1b[31mtest\x1b[0m") == "test"
def test_public_api_exports(self):
"""All expected exports should be available."""
from task_logger import (
LogPhase,
LogEntryType,
LogEntry,
TaskLogger,
load_task_logs,
get_active_phase,
get_task_logger,
clear_task_logger,
update_task_logger_path,
strip_ansi_codes,
StreamingLogCapture,
)
# If imports succeed, the test passes