auto-sync: tudo-para-ia-mais-humana 2026-05-02 07:28:45
This commit is contained in:
@@ -52,6 +52,7 @@ from .scanner import environment_summary, scan_ecosystem
|
||||
from .storage import table_counts
|
||||
from .targeted_sync_audit import run_targeted_sync_audit
|
||||
from .workspace_hygiene import run_workspace_hygiene
|
||||
from .workspace_hygiene_policy import policy_payload, run_hygiene_policy
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
@@ -126,6 +127,10 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
hygiene.add_argument("--project-root", default="G:/_codex-git/tudo-para-ia-mais-humana")
|
||||
hygiene.add_argument("--central-platform-folder", default="")
|
||||
hygiene.add_argument("--apply", action="store_true")
|
||||
hygiene_policy = sub.add_parser("workspace-hygiene-policy", help="Write executable workspace cleanup and ACL retention policy.")
|
||||
hygiene_policy.add_argument("--project-root", default="G:/_codex-git/tudo-para-ia-mais-humana")
|
||||
hygiene_policy.add_argument("--no-generated", action="store_true")
|
||||
hygiene_policy.add_argument("--limit", type=int, default=40)
|
||||
sync_audit = sub.add_parser("targeted-sync-audit", help="Write safe Git synchronization audit for the active round repos.")
|
||||
sync_audit.add_argument("--project-root", default="G:/_codex-git/tudo-para-ia-mais-humana")
|
||||
sync_audit.add_argument("--mcp-repo-root", default="G:/_codex-git/tudo-para-ia-mcps-internos-plataform")
|
||||
@@ -557,6 +562,19 @@ def command_workspace_hygiene(args: argparse.Namespace) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
def command_workspace_hygiene_policy(args: argparse.Namespace) -> int:
|
||||
report, records = run_hygiene_policy(
|
||||
project_root=Path(args.project_root),
|
||||
use_generated=not bool(args.no_generated),
|
||||
)
|
||||
payload = {
|
||||
"report": policy_payload(report, limit_cases=int(args.limit)),
|
||||
"generatedFiles": [record.path for record in records],
|
||||
}
|
||||
print(json.dumps(payload, ensure_ascii=False, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
def command_targeted_sync_audit(args: argparse.Namespace) -> int:
|
||||
central_platform_folder = Path(args.central_platform_folder) if args.central_platform_folder else None
|
||||
report, records = run_targeted_sync_audit(
|
||||
@@ -666,6 +684,8 @@ def main(argv: list[str] | None = None) -> int:
|
||||
return command_mcp_access_policy(args)
|
||||
if args.command == "workspace-hygiene":
|
||||
return command_workspace_hygiene(args)
|
||||
if args.command == "workspace-hygiene-policy":
|
||||
return command_workspace_hygiene_policy(args)
|
||||
if args.command == "targeted-sync-audit":
|
||||
return command_targeted_sync_audit(args)
|
||||
if args.command == "mcp-admin-route-acceptance":
|
||||
|
||||
38715
src/mais_humana/generated_workspace_hygiene_policy.py
Normal file
38715
src/mais_humana/generated_workspace_hygiene_policy.py
Normal file
File diff suppressed because it is too large
Load Diff
@@ -13,7 +13,6 @@ import csv
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
@@ -196,13 +195,61 @@ def _footprint(path: Path, *, max_errors: int = 8) -> PathFootprint:
|
||||
return PathFootprint(True, True, False, child_count, byte_count, tuple(errors[:max_errors]))
|
||||
|
||||
|
||||
def _delete_file(path: Path, errors: list[str], *, max_errors: int = 12) -> None:
|
||||
try:
|
||||
if not path.exists():
|
||||
return
|
||||
path.unlink()
|
||||
except OSError as exc:
|
||||
if len(errors) < max_errors:
|
||||
errors.append(f"{type(exc).__name__}: {exc}")
|
||||
|
||||
|
||||
def _delete_empty_dir(path: Path, errors: list[str], *, max_errors: int = 12) -> None:
|
||||
try:
|
||||
if path.exists():
|
||||
path.rmdir()
|
||||
except OSError as exc:
|
||||
if len(errors) < max_errors:
|
||||
errors.append(f"{type(exc).__name__}: {exc}")
|
||||
|
||||
|
||||
def _delete_directory_best_effort(path: Path) -> tuple[bool, str]:
|
||||
"""Delete a directory while continuing after ACL errors.
|
||||
|
||||
``shutil.rmtree`` stops at the first inaccessible child on Windows. The
|
||||
closeout hygiene needs a better operational behavior: remove every
|
||||
accessible file/directory, retain only the ACL-blocked paths, and report the
|
||||
retained evidence.
|
||||
"""
|
||||
|
||||
errors: list[str] = []
|
||||
|
||||
def on_walk_error(exc: OSError) -> None:
|
||||
if len(errors) < 12:
|
||||
errors.append(f"{type(exc).__name__}: {exc}")
|
||||
|
||||
for current_root, dirnames, filenames in os.walk(path, topdown=False, onerror=on_walk_error):
|
||||
current = Path(current_root)
|
||||
for filename in filenames:
|
||||
_delete_file(current / filename, errors)
|
||||
for dirname in dirnames:
|
||||
_delete_empty_dir(current / dirname, errors)
|
||||
_delete_empty_dir(path, errors)
|
||||
if not path.exists():
|
||||
return True, ""
|
||||
message = "; ".join(merge_unique(errors))
|
||||
if not message:
|
||||
message = "path retained after best-effort cleanup"
|
||||
return False, message
|
||||
|
||||
|
||||
def _delete_path(path: Path) -> tuple[bool, str]:
|
||||
try:
|
||||
if not path.exists():
|
||||
return False, ""
|
||||
if path.is_dir():
|
||||
shutil.rmtree(path)
|
||||
return True, ""
|
||||
return _delete_directory_best_effort(path)
|
||||
path.unlink()
|
||||
return True, ""
|
||||
except OSError as exc:
|
||||
|
||||
470
src/mais_humana/workspace_hygiene_policy.py
Normal file
470
src/mais_humana/workspace_hygiene_policy.py
Normal file
@@ -0,0 +1,470 @@
|
||||
"""Workspace hygiene retention policy.
|
||||
|
||||
The closeout rule is simple to say and surprisingly easy to get wrong on
|
||||
Windows mirrors: remove local build/test artifacts, but keep going when ACLs
|
||||
retain a child path and record exactly what still needs an authorized owner.
|
||||
This module turns that rule into executable policy cases that can be exposed to
|
||||
MCP, tested by Codex, and attached to the service-order closeout.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import io
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable, Sequence
|
||||
|
||||
from .models import GeneratedFile, as_plain_data, merge_unique, utc_now
|
||||
from .repository_mesh import stable_digest
|
||||
from .storage import connect, upsert_files
|
||||
|
||||
|
||||
class HygieneArtifactKind(str, Enum):
|
||||
"""Artifacts governed by the cleanup policy."""
|
||||
|
||||
PYTHON_TEST_TEMP = "python_test_temp"
|
||||
NODE_DEPENDENCIES = "node_dependencies"
|
||||
PYTHON_CACHE = "python_cache"
|
||||
WRANGLER_CACHE = "wrangler_cache"
|
||||
FRONTEND_CACHE = "frontend_cache"
|
||||
COVERAGE_OUTPUT = "coverage_output"
|
||||
TEST_REPORT = "test_report"
|
||||
LOG_OUTPUT = "log_output"
|
||||
|
||||
|
||||
class HygieneErrorKind(str, Enum):
|
||||
"""Filesystem or operational issue observed during cleanup."""
|
||||
|
||||
NONE = "none"
|
||||
NOT_FOUND = "not_found"
|
||||
PERMISSION_DENIED = "permission_denied"
|
||||
DIRECTORY_NOT_EMPTY = "directory_not_empty"
|
||||
FILE_LOCKED = "file_locked"
|
||||
UNSAFE_PATH = "unsafe_path"
|
||||
CENTRAL_WRITE_DENIED = "central_write_denied"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
class HygieneExecutionMode(str, Enum):
|
||||
"""Execution mode for a cleanup case."""
|
||||
|
||||
INSPECT = "inspect"
|
||||
APPLY = "apply"
|
||||
CLOSEOUT = "closeout"
|
||||
CENTRAL_RECORD = "central_record"
|
||||
|
||||
|
||||
class HygieneEnvironment(str, Enum):
|
||||
"""Environment where a cleanup case can run."""
|
||||
|
||||
WINDOWS_PRIMARY = "windows_primary"
|
||||
WINDOWS_SECONDARY = "windows_secondary"
|
||||
CODEX_SERVER = "codex_server"
|
||||
GITLAB_SERVER = "gitlab_server"
|
||||
|
||||
|
||||
class HygienePolicyStatus(str, Enum):
|
||||
"""Decision for one hygiene policy case."""
|
||||
|
||||
PASS = "pass"
|
||||
RETAIN_WITH_EVIDENCE = "retain_with_evidence"
|
||||
OWNER_ACTION_REQUIRED = "owner_action_required"
|
||||
BLOCK_UNSAFE = "block_unsafe"
|
||||
RECORD_ONLY = "record_only"
|
||||
|
||||
|
||||
class HygieneRemediationAction(str, Enum):
|
||||
"""Action recommended by a policy case."""
|
||||
|
||||
VERIFY_ABSENT = "verify_absent"
|
||||
BEST_EFFORT_DELETE = "best_effort_delete"
|
||||
CONTINUE_AFTER_CHILD_ERROR = "continue_after_child_error"
|
||||
ESCALATE_OWNER = "escalate_owner"
|
||||
RECORD_ACL_EXCEPTION = "record_acl_exception"
|
||||
BLOCK_BEFORE_WRITE = "block_before_write"
|
||||
WRITE_STATUS_ARTIFACT = "write_status_artifact"
|
||||
UPDATE_SEMANTIC_SQL = "update_semantic_sql"
|
||||
|
||||
|
||||
ARTIFACT_TARGETS: tuple[tuple[HygieneArtifactKind, str, str], ...] = (
|
||||
(HygieneArtifactKind.PYTHON_TEST_TEMP, ".test-tmp", "scratch criado por testes Python e comandos de validacao"),
|
||||
(HygieneArtifactKind.NODE_DEPENDENCIES, "node_modules", "dependencias Node locais que nao devem sobreviver ao fechamento"),
|
||||
(HygieneArtifactKind.PYTHON_CACHE, "__pycache__", "cache Python recompilavel"),
|
||||
(HygieneArtifactKind.PYTHON_CACHE, ".pytest_cache", "cache pytest local"),
|
||||
(HygieneArtifactKind.WRANGLER_CACHE, ".wrangler", "cache local do Wrangler"),
|
||||
(HygieneArtifactKind.FRONTEND_CACHE, ".next", "build cache frontend"),
|
||||
(HygieneArtifactKind.FRONTEND_CACHE, ".vite", "build cache frontend"),
|
||||
(HygieneArtifactKind.COVERAGE_OUTPUT, "coverage", "saida local de cobertura"),
|
||||
(HygieneArtifactKind.TEST_REPORT, "test-results", "saida local de testes"),
|
||||
(HygieneArtifactKind.TEST_REPORT, "playwright-report", "relatorio local de browser tests"),
|
||||
(HygieneArtifactKind.TEST_REPORT, "blob-report", "artefato local de browser tests"),
|
||||
(HygieneArtifactKind.LOG_OUTPUT, "logs", "logs locais de execucao"),
|
||||
)
|
||||
|
||||
POLICY_TRANSIT_FIELDS: tuple[str, ...] = (
|
||||
"origin",
|
||||
"destination",
|
||||
"tool",
|
||||
"payload",
|
||||
"actor",
|
||||
"permission",
|
||||
"result",
|
||||
"traceId",
|
||||
"auditId",
|
||||
"timestamp",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class HygienePolicyCase:
|
||||
"""One policy case for local artifact cleanup."""
|
||||
|
||||
case_id: str
|
||||
artifact_kind: HygieneArtifactKind
|
||||
relative_path: str
|
||||
environment: HygieneEnvironment
|
||||
execution_mode: HygieneExecutionMode
|
||||
error_kind: HygieneErrorKind
|
||||
status: HygienePolicyStatus
|
||||
remediation_actions: tuple[HygieneRemediationAction, ...]
|
||||
required_evidence: tuple[str, ...]
|
||||
mcp_transit_required: bool
|
||||
direct_delete_allowed: bool
|
||||
reason: str
|
||||
next_action: str
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
return as_plain_data(self)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class HygienePolicyReport:
|
||||
"""Policy report used by CLI, Markdown, CSV, and tests."""
|
||||
|
||||
report_id: str
|
||||
generated_at: str
|
||||
cases: tuple[HygienePolicyCase, ...]
|
||||
summary: tuple[str, ...]
|
||||
|
||||
@property
|
||||
def cases_count(self) -> int:
|
||||
return len(self.cases)
|
||||
|
||||
@property
|
||||
def owner_action_count(self) -> int:
|
||||
return sum(1 for case in self.cases if case.status == HygienePolicyStatus.OWNER_ACTION_REQUIRED)
|
||||
|
||||
@property
|
||||
def unsafe_block_count(self) -> int:
|
||||
return sum(1 for case in self.cases if case.status == HygienePolicyStatus.BLOCK_UNSAFE)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
data = as_plain_data(self)
|
||||
data["cases_count"] = self.cases_count
|
||||
data["owner_action_count"] = self.owner_action_count
|
||||
data["unsafe_block_count"] = self.unsafe_block_count
|
||||
return data
|
||||
|
||||
|
||||
def classify_policy_case(
|
||||
*,
|
||||
artifact_kind: HygieneArtifactKind,
|
||||
relative_path: str,
|
||||
environment: HygieneEnvironment,
|
||||
execution_mode: HygieneExecutionMode,
|
||||
error_kind: HygieneErrorKind,
|
||||
) -> tuple[HygienePolicyStatus, tuple[HygieneRemediationAction, ...], bool, str, str]:
|
||||
"""Classify one cleanup situation into an auditable policy decision."""
|
||||
|
||||
if error_kind == HygieneErrorKind.UNSAFE_PATH:
|
||||
return (
|
||||
HygienePolicyStatus.BLOCK_UNSAFE,
|
||||
(HygieneRemediationAction.BLOCK_BEFORE_WRITE, HygieneRemediationAction.WRITE_STATUS_ARTIFACT),
|
||||
False,
|
||||
"path calculado sai do project_root ou cruza boundary de volume",
|
||||
"bloquear antes de qualquer escrita e registrar evidencia",
|
||||
)
|
||||
if execution_mode == HygieneExecutionMode.INSPECT:
|
||||
return (
|
||||
HygienePolicyStatus.RECORD_ONLY,
|
||||
(HygieneRemediationAction.VERIFY_ABSENT, HygieneRemediationAction.WRITE_STATUS_ARTIFACT),
|
||||
False,
|
||||
"modo inspecao nunca apaga artefato",
|
||||
"registrar footprint e executar apply apenas no fechamento",
|
||||
)
|
||||
if error_kind in {HygieneErrorKind.NONE, HygieneErrorKind.NOT_FOUND}:
|
||||
return (
|
||||
HygienePolicyStatus.PASS,
|
||||
(HygieneRemediationAction.VERIFY_ABSENT,),
|
||||
True,
|
||||
"artefato ausente ou removivel por limpeza local aprovada",
|
||||
"manter ausente e registrar status",
|
||||
)
|
||||
if error_kind in {HygieneErrorKind.PERMISSION_DENIED, HygieneErrorKind.FILE_LOCKED}:
|
||||
return (
|
||||
HygienePolicyStatus.OWNER_ACTION_REQUIRED,
|
||||
(
|
||||
HygieneRemediationAction.BEST_EFFORT_DELETE,
|
||||
HygieneRemediationAction.CONTINUE_AFTER_CHILD_ERROR,
|
||||
HygieneRemediationAction.ESCALATE_OWNER,
|
||||
HygieneRemediationAction.RECORD_ACL_EXCEPTION,
|
||||
),
|
||||
True,
|
||||
"ACL ou processo externo reteve caminho depois de remover filhos acessiveis",
|
||||
"registrar caminho retido e solicitar owner autorizado",
|
||||
)
|
||||
if error_kind == HygieneErrorKind.DIRECTORY_NOT_EMPTY:
|
||||
return (
|
||||
HygienePolicyStatus.RETAIN_WITH_EVIDENCE,
|
||||
(
|
||||
HygieneRemediationAction.BEST_EFFORT_DELETE,
|
||||
HygieneRemediationAction.CONTINUE_AFTER_CHILD_ERROR,
|
||||
HygieneRemediationAction.WRITE_STATUS_ARTIFACT,
|
||||
),
|
||||
True,
|
||||
"diretorio ainda contem filho retido por erro anterior",
|
||||
"reduzir conteudo acessivel e registrar filhos retidos",
|
||||
)
|
||||
if error_kind == HygieneErrorKind.CENTRAL_WRITE_DENIED:
|
||||
return (
|
||||
HygienePolicyStatus.RETAIN_WITH_EVIDENCE,
|
||||
(HygieneRemediationAction.WRITE_STATUS_ARTIFACT, HygieneRemediationAction.UPDATE_SEMANTIC_SQL),
|
||||
False,
|
||||
"pasta central recusou escrita, mas projeto real deve registrar status",
|
||||
"manter fallback no projeto real e registrar pendencia central",
|
||||
)
|
||||
return (
|
||||
HygienePolicyStatus.RETAIN_WITH_EVIDENCE,
|
||||
(HygieneRemediationAction.WRITE_STATUS_ARTIFACT,),
|
||||
False,
|
||||
"erro desconhecido deve ser evidenciado antes de nova tentativa",
|
||||
"registrar erro bruto redigido e reavaliar proxima rodada",
|
||||
)
|
||||
|
||||
|
||||
def build_policy_cases() -> tuple[HygienePolicyCase, ...]:
|
||||
"""Build the deterministic matrix of hygiene policy cases."""
|
||||
|
||||
cases: list[HygienePolicyCase] = []
|
||||
for artifact_kind, relative_path, description in ARTIFACT_TARGETS:
|
||||
for environment in HygieneEnvironment:
|
||||
for execution_mode in HygieneExecutionMode:
|
||||
for error_kind in HygieneErrorKind:
|
||||
status, actions, direct_delete_allowed, reason, next_action = classify_policy_case(
|
||||
artifact_kind=artifact_kind,
|
||||
relative_path=relative_path,
|
||||
environment=environment,
|
||||
execution_mode=execution_mode,
|
||||
error_kind=error_kind,
|
||||
)
|
||||
seed = {
|
||||
"artifact": artifact_kind.value,
|
||||
"path": relative_path,
|
||||
"environment": environment.value,
|
||||
"mode": execution_mode.value,
|
||||
"error": error_kind.value,
|
||||
}
|
||||
evidence = (
|
||||
"footprint_before",
|
||||
"footprint_after",
|
||||
"git_status_short",
|
||||
"node_modules_absent",
|
||||
"acl_error_excerpt" if error_kind != HygieneErrorKind.NONE else "no_error",
|
||||
description,
|
||||
)
|
||||
cases.append(
|
||||
HygienePolicyCase(
|
||||
case_id=f"workspace-hygiene-policy-{stable_digest(seed, 24)}",
|
||||
artifact_kind=artifact_kind,
|
||||
relative_path=relative_path,
|
||||
environment=environment,
|
||||
execution_mode=execution_mode,
|
||||
error_kind=error_kind,
|
||||
status=status,
|
||||
remediation_actions=actions,
|
||||
required_evidence=evidence,
|
||||
mcp_transit_required=True,
|
||||
direct_delete_allowed=direct_delete_allowed,
|
||||
reason=reason,
|
||||
next_action=next_action,
|
||||
)
|
||||
)
|
||||
return tuple(cases)
|
||||
|
||||
|
||||
def build_hygiene_policy_cases(*, use_generated: bool = True) -> tuple[HygienePolicyCase, ...]:
|
||||
"""Return generated policy cases when available, otherwise build runtime cases."""
|
||||
|
||||
if use_generated:
|
||||
try:
|
||||
from .generated_workspace_hygiene_policy import iter_policy_cases
|
||||
|
||||
return tuple(iter_policy_cases())
|
||||
except (ImportError, AttributeError):
|
||||
pass
|
||||
return build_policy_cases()
|
||||
|
||||
|
||||
def build_hygiene_policy_report(*, use_generated: bool = True) -> HygienePolicyReport:
|
||||
"""Build a compact policy report."""
|
||||
|
||||
cases = build_hygiene_policy_cases(use_generated=use_generated)
|
||||
summary = (
|
||||
f"Policy cases: {len(cases)}.",
|
||||
f"Owner action cases: {sum(1 for case in cases if case.status == HygienePolicyStatus.OWNER_ACTION_REQUIRED)}.",
|
||||
f"Unsafe path blocks: {sum(1 for case in cases if case.status == HygienePolicyStatus.BLOCK_UNSAFE)}.",
|
||||
"Best-effort cleanup must continue after child ACL errors and retain only inaccessible paths.",
|
||||
)
|
||||
return HygienePolicyReport(
|
||||
report_id=f"workspace-hygiene-policy-{stable_digest([case.case_id for case in cases], 16)}",
|
||||
generated_at=utc_now(),
|
||||
cases=cases,
|
||||
summary=summary,
|
||||
)
|
||||
|
||||
|
||||
def policy_payload(report: HygienePolicyReport, *, limit_cases: int = 80) -> dict[str, Any]:
|
||||
"""Return compact JSON payload for the policy report."""
|
||||
|
||||
payload = report.to_dict()
|
||||
payload["transit_fields"] = list(POLICY_TRANSIT_FIELDS)
|
||||
payload["cases"] = [case.to_dict() for case in report.cases[: max(0, limit_cases)]]
|
||||
payload["cases_total"] = len(report.cases)
|
||||
return payload
|
||||
|
||||
|
||||
def policy_case_rows(cases: Sequence[HygienePolicyCase]) -> list[list[str]]:
|
||||
"""Return CSV rows for policy cases."""
|
||||
|
||||
rows = [
|
||||
[
|
||||
"case_id",
|
||||
"artifact_kind",
|
||||
"relative_path",
|
||||
"environment",
|
||||
"execution_mode",
|
||||
"error_kind",
|
||||
"status",
|
||||
"remediation_actions",
|
||||
"direct_delete_allowed",
|
||||
"reason",
|
||||
"next_action",
|
||||
]
|
||||
]
|
||||
for case in cases:
|
||||
rows.append(
|
||||
[
|
||||
case.case_id,
|
||||
case.artifact_kind.value,
|
||||
case.relative_path,
|
||||
case.environment.value,
|
||||
case.execution_mode.value,
|
||||
case.error_kind.value,
|
||||
case.status.value,
|
||||
" | ".join(action.value for action in case.remediation_actions),
|
||||
"yes" if case.direct_delete_allowed else "no",
|
||||
case.reason,
|
||||
case.next_action,
|
||||
]
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def rows_to_csv(rows: Sequence[Sequence[str]]) -> str:
|
||||
"""Serialize rows to CSV text."""
|
||||
|
||||
buffer = io.StringIO()
|
||||
writer = csv.writer(buffer, lineterminator="\n")
|
||||
writer.writerows(rows)
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
def policy_markdown(report: HygienePolicyReport, *, limit_cases: int = 40) -> str:
|
||||
"""Render the policy report in Markdown."""
|
||||
|
||||
lines = [
|
||||
"# Politica de higiene de workspace",
|
||||
"",
|
||||
f"- report_id: `{report.report_id}`",
|
||||
f"- generated_at: `{report.generated_at}`",
|
||||
f"- cases: `{report.cases_count}`",
|
||||
f"- owner_action_cases: `{report.owner_action_count}`",
|
||||
f"- unsafe_block_cases: `{report.unsafe_block_count}`",
|
||||
"",
|
||||
"## Resumo",
|
||||
"",
|
||||
]
|
||||
lines.extend(f"- {item}" for item in report.summary)
|
||||
lines.extend(["", "## Casos amostrais", ""])
|
||||
for case in report.cases[: max(0, limit_cases)]:
|
||||
actions = ", ".join(action.value for action in case.remediation_actions)
|
||||
lines.append(
|
||||
f"- `{case.case_id}` `{case.relative_path}` `{case.environment.value}` `{case.execution_mode.value}` "
|
||||
f"`{case.error_kind.value}` -> `{case.status.value}` actions `{actions}`"
|
||||
)
|
||||
lines.extend(
|
||||
[
|
||||
"",
|
||||
"## Regra operacional",
|
||||
"",
|
||||
"- Remover somente artefatos locais aprovados.",
|
||||
"- Continuar a limpeza depois de erro de filho quando o caminho ainda estiver dentro do projeto.",
|
||||
"- Registrar WinError 5, arquivo em uso ou pasta nao vazia como pendencia de owner quando persistirem.",
|
||||
"- Nao apagar paths fora do project_root.",
|
||||
"",
|
||||
]
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def generated_files(project_root: Path) -> tuple[GeneratedFile, ...]:
|
||||
"""Return semantic records for policy artifacts."""
|
||||
|
||||
relation = "0041_EXECUTIVA__resolver-test-tmp-bloqueado-por-acl"
|
||||
specs = (
|
||||
("dados/workspace-hygiene-policy.json", "Politica JSON de retencao e limpeza de workspace.", "workspace hygiene policy", "json"),
|
||||
("matrizes/workspace-hygiene-policy-cases.csv", "Matriz de casos de higiene e ACL.", "workspace hygiene policy cases", "csv"),
|
||||
("ecossistema/WORKSPACE-HYGIENE-POLICY.md", "Relatorio humano da politica de higiene.", "workspace hygiene policy report", "markdown"),
|
||||
)
|
||||
return tuple(
|
||||
GeneratedFile(
|
||||
path=str(project_root / relative),
|
||||
description=description,
|
||||
function=function,
|
||||
file_type=file_type,
|
||||
changed_by="mais_humana.workspace_hygiene_policy",
|
||||
change_summary="Criada politica executavel para limpeza parcial, ACL, owner action e evidencia de fechamento.",
|
||||
relation_to_order=relation,
|
||||
)
|
||||
for relative, description, function, file_type in specs
|
||||
)
|
||||
|
||||
|
||||
def write_policy_artifacts(report: HygienePolicyReport, project_root: Path) -> tuple[GeneratedFile, ...]:
|
||||
"""Write policy JSON, CSV, Markdown, and semantic file records."""
|
||||
|
||||
targets = (
|
||||
(project_root / "dados" / "workspace-hygiene-policy.json", json.dumps(policy_payload(report, limit_cases=160), ensure_ascii=False, indent=2, sort_keys=True)),
|
||||
(project_root / "matrizes" / "workspace-hygiene-policy-cases.csv", rows_to_csv(policy_case_rows(report.cases))),
|
||||
(project_root / "ecossistema" / "WORKSPACE-HYGIENE-POLICY.md", policy_markdown(report, limit_cases=80)),
|
||||
)
|
||||
for path, text in targets:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(text, encoding="utf-8")
|
||||
records = generated_files(project_root)
|
||||
with connect(project_root / "controle-semantico.sqlite") as conn:
|
||||
upsert_files(conn, records)
|
||||
conn.commit()
|
||||
return records
|
||||
|
||||
|
||||
def run_hygiene_policy(project_root: Path, *, use_generated: bool = True) -> tuple[HygienePolicyReport, tuple[GeneratedFile, ...]]:
|
||||
"""Build and persist the workspace hygiene policy artifacts."""
|
||||
|
||||
report = build_hygiene_policy_report(use_generated=use_generated)
|
||||
records = write_policy_artifacts(report, project_root)
|
||||
return report, records
|
||||
|
||||
Reference in New Issue
Block a user