feat(T06): hostPath working-memory sync and repo-relative sink paths
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 3s
Build and Publish Container Image / build-and-push (push) Successful in 1m36s

Mount the-custodian memory/working from the railiance01 clone (hostPath)
so sweep writeback commits daily-triage notes; worker runs as uid 1000;
progress events store repo-relative working_memory_path; ops inventory
defaults use custodian:// URIs.
This commit is contained in:
tegwick 2026-07-07 01:00:49 +02:00
parent 70e3154b05
commit ce03e78e26
10 changed files with 76 additions and 30 deletions

View file

@ -24,7 +24,7 @@ from activity_core.sync_activity_definitions import ACTIVITY_DEFINITION_ID_NAMES
DEFAULT_TIMEZONE = "Europe/Berlin"
DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000"
DEFAULT_WORKING_MEMORY_DIR = "/home/worsch/the-custodian/memory/working"
from activity_core.runtime_paths import custodian_working_memory_dir
DEFAULT_TEMPORAL_NAMESPACE = "default"
FAILURE_STATUSES = {"missed", "validation_failed", "sink_failed"}
WEEKDAYS = {
@ -57,7 +57,13 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser.add_argument("--activity-name", action="append", default=[])
parser.add_argument("--db-url", default=os.environ.get("ACTCORE_DB_URL"))
parser.add_argument("--state-hub-url", default=os.environ.get("STATE_HUB_URL", DEFAULT_STATE_HUB_URL))
parser.add_argument("--working-memory-dir", default=os.environ.get("AUTOMATION_STATUS_WORKING_MEMORY_DIR", DEFAULT_WORKING_MEMORY_DIR))
parser.add_argument(
"--working-memory-dir",
default=os.environ.get(
"AUTOMATION_STATUS_WORKING_MEMORY_DIR",
str(custodian_working_memory_dir()),
),
)
parser.add_argument("--temporal-host", default=os.environ.get("TEMPORAL_HOST"))
parser.add_argument("--temporal-namespace", default=os.environ.get("TEMPORAL_NAMESPACE", DEFAULT_TEMPORAL_NAMESPACE))
parser.add_argument("--timeout-seconds", type=float, default=float(os.environ.get("AUTOMATION_STATUS_TIMEOUT_SECONDS", "5")))

View file

@ -20,8 +20,7 @@ import httpx
import yaml
from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY, ContextResolver
_DEFAULT_INVENTORY_PATH = "/home/worsch/the-custodian/ops/service-inventory.yml"
from activity_core.runtime_paths import default_ops_inventory_path
_DEFAULT_TIMEOUT_SECONDS = 10.0
_SUPPORTED_ENDPOINT_TYPES = {"http", "https"}
@ -39,12 +38,11 @@ CONTEXT_RESOLVER_REGISTRY["ops-inventory"] = OpsInventoryContextResolver
def _probe_services(params: dict[str, Any]) -> dict[str, Any]:
inventory_path = Path(
str(
params.get("inventory_path")
or os.environ.get("OPS_INVENTORY_PATH")
or _DEFAULT_INVENTORY_PATH
)
inventory_raw = params.get("inventory_path")
inventory_path = (
default_ops_inventory_path()
if not inventory_raw
else Path(str(inventory_raw)).expanduser()
)
timeout_seconds = float(params.get("timeout_seconds", _DEFAULT_TIMEOUT_SECONDS))
allow_network = _bool_param(params.get("allow_network", True))

View file

@ -11,7 +11,11 @@ from zoneinfo import ZoneInfo
import httpx
from activity_core.runtime_paths import custodian_repo_root, resolve_runtime_path
from activity_core.runtime_paths import (
custodian_repo_relative,
custodian_repo_root,
resolve_runtime_path,
)
from activity_core.state_hub_write import idempotency_headers
_DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000"
@ -82,13 +86,14 @@ def _write_working_memory(
target = (directory / filename).resolve()
_assert_allowed_output_path(target)
repo_relative = custodian_repo_relative(target)
if target.exists():
text = target.read_text(encoding="utf-8")
if f"activity_core_run_id: {run_id}" in text:
return {
"type": "working-memory",
"status": "exists",
"path": str(target),
"path": repo_relative,
}
raise FileExistsError(f"refusing to overwrite existing report note: {target}")
@ -97,7 +102,7 @@ def _write_working_memory(
return {
"type": "working-memory",
"status": "written",
"path": str(target),
"path": repo_relative,
}

View file

@ -34,4 +34,24 @@ def resolve_runtime_path(raw_path: str) -> Path:
if value.startswith(_ACTIVITY_CORE_SCHEME):
return (activity_core_root() / value.removeprefix(_ACTIVITY_CORE_SCHEME)).resolve()
return Path(value).expanduser()
return Path(value).expanduser()
def custodian_working_memory_dir() -> Path:
return resolve_runtime_path("custodian://memory/working")
def custodian_repo_relative(path: Path) -> str:
"""Return a repo-relative POSIX path when under CUSTODIAN_REPO_ROOT."""
root = custodian_repo_root().resolve()
try:
return path.resolve().relative_to(root).as_posix()
except ValueError:
return path.as_posix()
def default_ops_inventory_path() -> Path:
raw = os.environ.get("OPS_INVENTORY_PATH", "").strip()
if raw:
return resolve_runtime_path(raw)
return resolve_runtime_path("custodian://ops/service-inventory.yml")