mirror of
https://github.com/yogeshojha/rengine.git
synced 2026-10-21 07:41:28 +02:00
Yogesh Ojha
2faf1dddaf
- Add aggregate functions for scan status and counts in orchestrator. - Create ScanEventPublisher for emitting scan-related events. - Introduce ScanActivityService for managing scan activities. - Implement ScanCommandRecorder for logging command executions. - Develop canvas building and finalization logic for scan tasks. - Refactor scan task to utilize orchestrator functions for running stages and finalizing scans. - Enhance scan notification system to dispatch relevant events. - Integrate command recording into CLI tool execution for better tracking.
49 lines
1.1 KiB
Python
49 lines
1.1 KiB
Python
"""Tool-agnostic models for CLI tool execution results."""
|
|
|
|
from enum import Enum
|
|
from typing import Any, Protocol, runtime_checkable
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class OutputFormat(Enum):
|
|
"""Output format expected from the CLI tool."""
|
|
|
|
JSONL = "jsonl"
|
|
PLAIN = "plain"
|
|
|
|
|
|
@runtime_checkable
|
|
class CommandRecorder(Protocol):
|
|
"""Optional hook the executor calls to register a command and capture its log."""
|
|
|
|
def start(self, tool: str, command: str) -> Any: ...
|
|
|
|
def finish(
|
|
self,
|
|
handle: Any,
|
|
*,
|
|
return_code: int,
|
|
output: str,
|
|
error: str | None,
|
|
duration_seconds: float,
|
|
) -> None: ...
|
|
|
|
|
|
class ToolResult(BaseModel):
|
|
"""Result of a CLI tool execution."""
|
|
|
|
success: bool
|
|
stdout: str = ""
|
|
stderr: str = ""
|
|
exit_code: int = -1
|
|
output_lines: list[str] = Field(default_factory=list)
|
|
json_records: list[dict] = Field(default_factory=list)
|
|
duration_seconds: float = 0.0
|
|
command: str = ""
|
|
error: str | None = None
|
|
|
|
@property
|
|
def has_output(self) -> bool:
|
|
return bool(self.json_records or self.output_lines)
|