diff --git a/.github/workflows/position-span-of-control-quality.yml b/.github/workflows/position-span-of-control-quality.yml new file mode 100644 index 000000000..869c2c925 --- /dev/null +++ b/.github/workflows/position-span-of-control-quality.yml @@ -0,0 +1,58 @@ +name: Position Span of Control Quality + +on: + pull_request: + branches: + - develop + - feat/position-reporting-hierarchy + paths: + - '.github/workflows/position-span-of-control-quality.yml' + - '.github/requirements/foundation-test.txt' + - 'packages/hris-kernel/**' + - 'docs/adr/0133-position-span-of-control-evidence.md' + - 'docs/doctoring/position-span-of-control-references.md' + - 'docs/traceability/position-span-of-control.md' + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: position-span-of-control-quality-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + span-of-control: + name: Position span-of-control contract and HRIS kernel 100% coverage + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - name: Checkout exact candidate + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Prove exact candidate checkout + env: + ORGMETRA_EXPECTED_HEAD_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: test "$(git rev-parse HEAD)" = "$ORGMETRA_EXPECTED_HEAD_SHA" + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: '3.14' + check-latest: false + - name: Install reviewed test toolchain + run: | + python -m pip install --require-hashes --no-deps --only-binary=:all: -r .github/requirements/foundation-test.txt + python -m pip check + - name: Compile HRIS workforce boundary + run: python -m compileall -q packages/hris-kernel/src packages/hris-kernel/tests + - name: Test complete HRIS kernel with exact statement and branch coverage + env: + PYTHONPATH: packages/hris-kernel/src + COVERAGE_FILE: /tmp/orgmetra-position-span.coverage + run: python -m pytest -c packages/hris-kernel/pyproject.toml packages/hris-kernel/tests + - name: Require clean checkout + run: | + git diff --exit-code + test -z "$(git status --porcelain)" diff --git a/docs/adr/0133-position-span-of-control-evidence.md b/docs/adr/0133-position-span-of-control-evidence.md new file mode 100644 index 000000000..61eb15a84 --- /dev/null +++ b/docs/adr/0133-position-span-of-control-evidence.md @@ -0,0 +1,39 @@ +# ADR 0133: Position span-of-control evidence is structural, bitemporal, and non-authorizing + +- Status: Proposed +- Scope: active PR #133 only; not protected-`develop` truth until integrated +- Parent owner: PR #94 Position-to-Position solid-line reporting hierarchy + +## Context + +Orgmetra can reconstruct a tenant-scoped solid-line Position reporting graph on PR #94, but buyers still need a defensible organization-design metric answering a narrower question: how many direct-report **Position seats** are attached to each manager Position at one business date and one system-knowledge cutoff? + +Span of control is empirically consequential, but research does not support treating one universal number as optimal across all organizations or levels. Work complexity, functional diversity, hierarchy level and organizational goals change the relationship. Encoding a universal target would turn descriptive structure into an unsupported management or employment recommendation. + +## Decision + +Orgmetra will expose `PositionSpanOfControlSnapshot` as PII-minimized structural workforce evidence derived only from one exact governed `PositionReportingSnapshot`. + +The boundary: + +1. counts direct-report Position seats, never workers; +2. preserves the parent's explicit `tenant_record_id`, business `effective_on`, and UTC system `known_at` coordinate; +3. revalidates the parent runtime type, opaque identities, immutable edge shape, unique visible relationship identities, one manager per subordinate, self-edge rejection, and cycle rejection so direct dataclass construction cannot forge downstream evidence; +4. emits manager Position UUID plus positive direct-report Position count in deterministic UUID order; +5. binds canonical JSON with SHA-256 for audit correlation while excluding Person, Employment and Assignment identifiers; +6. labels the result `structural_workforce_evidence` and `not_authorized_for_employment_decision`; and +7. does **not** encode an ideal span, performance score, staffing recommendation, promotion signal, termination signal, or compensation implication. + +## Consequences + +This slice makes span-of-control structure inspectable without claiming that a wide or narrow span is good or bad. Any future recommendation layer must supply separately governed contextual evidence and human review rather than converting this descriptive count into a high-impact rule. + +Persistence, authorized presentation, organization-design simulation and workflow-specific UI remain separate boundaries. PR #94 must integrate first; this child is not shipped truth and cannot inherit parent checks or reviews. + +## References + +Bell, G. D. (1967). Determinants of span of control. *American Journal of Sociology, 73*(1), 90–101. https://doi.org/10.1086/224439 + +Meier, K. J., & Bohte, J. (2000). Ode to Luther Gulick: Span of control and organizational performance. *Administration & Society, 32*(2), 115–137. https://doi.org/10.1177/00953990022019371 + +Theobald, N. A., & Nicholson-Crotty, S. (2005). The many faces of span of control: Organizational structure across multiple goals. *Administration & Society, 36*(6), 648–660. https://doi.org/10.1177/0095399704270585 diff --git a/docs/doctoring/position-span-of-control-references.md b/docs/doctoring/position-span-of-control-references.md new file mode 100644 index 000000000..a3f56d8df --- /dev/null +++ b/docs/doctoring/position-span-of-control-references.md @@ -0,0 +1,23 @@ +# Position span-of-control research notes + +Retrieved for PR #133 on 2026-08-28. These references support a **descriptive structural metric**, not a universal managerial target. + +## Evidence used + +Bell (1967) examined supervisors and employees in a community hospital and reported that subordinate and supervisor job complexity were associated with narrower spans. Orgmetra therefore does not translate a direct-report count into a context-free judgement. + +Meier and Bohte (2000) tested span-of-control relationships using 678 Texas school districts over four years and found meaningful associations with organizational performance. The finding supports preserving the structural measure as evidence while not treating it as causal or universally optimal outside that design. + +Theobald and Nicholson-Crotty (2005) analyzed nearly 600 public bureaucracies and showed that structural choices can have different consequences across multiple organizational goals. Orgmetra therefore records the direct-report Position count and leaves goal-dependent interpretation to a separately governed, human-reviewed decision process. + +## Implementation constraint + +`PositionSpanOfControlSnapshot` counts subordinate **Position seats** under a manager Position at one tenant/business/system coordinate. It excludes worker identity and does not recommend a target span, employment action, compensation action, performance rating or organization change. + +## APA 7 references + +Bell, G. D. (1967). Determinants of span of control. *American Journal of Sociology, 73*(1), 90–101. https://doi.org/10.1086/224439 + +Meier, K. J., & Bohte, J. (2000). Ode to Luther Gulick: Span of control and organizational performance. *Administration & Society, 32*(2), 115–137. https://doi.org/10.1177/00953990022019371 + +Theobald, N. A., & Nicholson-Crotty, S. (2005). The many faces of span of control: Organizational structure across multiple goals. *Administration & Society, 36*(6), 648–660. https://doi.org/10.1177/0095399704270585 diff --git a/docs/traceability/position-span-of-control.md b/docs/traceability/position-span-of-control.md new file mode 100644 index 000000000..7072cc1c6 --- /dev/null +++ b/docs/traceability/position-span-of-control.md @@ -0,0 +1,27 @@ +# Position span-of-control traceability + +## State classification + +- Protected `develop`: does **not** contain Position-to-Position reporting or span-of-control evidence at `9e3e4847510e1e612b48474ba42b177b8ed824df`. +- Parent active PR #94: owns governed bitemporal Position reporting at `3f67182bb3065f2fc8fd974bfdd75a390d8a8fdc`. +- Active child PR #133: owns descriptive Position span-of-control evidence. It is dependency-first active-PR truth only. +- Planned/out of scope here: persistence, authorized buyer presentation, organization-design simulation, target-span recommendations, staffing actions and employment decisions. + +## Requirements to evidence + +| Requirement | Implementation | Regression / evidence | +|---|---|---| +| Count Position seats, not workers | `build_position_span_of_control_snapshot()` consumes only `PositionReportingSnapshot` | `test_span_snapshot_counts_direct_reporting_positions_only` and canonical JSON PII assertions | +| Preserve bitemporal coordinate | tenant/effective/system fields copied from exact parent snapshot | noncanonical parent-coordinate regressions | +| Resist forged direct parent construction | exact runtime, UUID, tuple, uniqueness, self-edge and cycle revalidation | hostile graph/container/runtime regressions | +| Deterministic structural evidence | UUID-sorted `span_by_manager`, exact positive counts, canonical JSON + SHA-256 | direct output and digest regressions | +| No universal target / no high-impact authority | fixed evidence state and decision-authority labels; no recommendation code path | public contract + ADR 0133 | +| Exact owned coverage | dedicated `Position Span of Control Quality` workflow runs the complete HRIS-kernel suite with existing exact 100% statement/branch thresholds | `.github/workflows/position-span-of-control-quality.yml` | + +## Dependency and integration rule + +PR #94 must integrate first. A focused GREEN result on #133 is stack-local only. After parent integration, retarget #133 to fresh `develop`, reconcile any parent changes, then rerun every applicable Workforce/People/Job-Analysis/Foundation/Recovery/SAST/Security and central required workflow on one new exact head. Parent checks, reviews and statuses never transfer. + +## Scientific interpretation guard + +Primary research shows span of control can matter but varies with work complexity, hierarchy and organizational goals. Orgmetra therefore exposes direct-report Position counts as inspectable structural evidence only. Any later evaluative or prescriptive use requires separate contextual evidence and accountable human review. diff --git a/packages/hris-kernel/README.md b/packages/hris-kernel/README.md index f7f768fc3..34212f14e 100644 --- a/packages/hris-kernel/README.md +++ b/packages/hris-kernel/README.md @@ -13,10 +13,13 @@ Use this package to: 7. Build a deterministic `JobAnalysisSnapshot` that links observable Tasks to explicit KSAO requirements, retains source/version/digest provenance, and optionally carries historical Functional Job Analysis Data/People/Things codes without confusing Job with Position or Assignment. 8. Build a deterministic `WorkforceCompositionSnapshot` for one tenant, effective day, and recorded-time cutoff with distinct-person headcount, reportable employment count, staffed assignment count/FTE, unassigned-person count, and status counts without serializing row-level person, employment, position, or assignment identifiers. 9. Build a deterministic solid-line `PositionReportingSnapshot` for one tenant, effective day, and system-knowledge cutoff without deriving supervision from Person, Assignment, or organization-unit parentage. +10. Build a deterministic `PositionSpanOfControlSnapshot` from one governed Position reporting snapshot to count direct-report Position seats per manager Position without counting workers or recommending an ideal span. Every historical reconstruction and portfolio/capacity decision requires an explicit `tenant_record_id`. A colliding durable identifier from another tenant is ignored rather than treated as local employment truth. -`PositionReportingRelationship` represents managerial structure between durable Position seats. Both endpoints must resolve to exactly one same-tenant `active` or `open` `PositionVersion` at the requested business/system coordinate. One subordinate can have only one visible solid-line manager, self-reporting and cycles fail closed, and caller-defined runtime subclasses cannot control identity or temporal comparisons. `build_position_reporting_snapshot(...)` returns deterministic subordinate-to-manager UUID pairs while its routine `repr` redacts those correlation identifiers. This is descriptive organizational evidence only: it neither identifies the worker occupying a seat nor grants employment-decision authority. Persistence and reporting-line mutation remain separate authoritative write-boundary work. +`PositionReportingRelationship` represents managerial structure between durable Position seats. Both endpoints must resolve to exactly one same-tenant `active` or `open` `PositionVersion` at the requested business/system coordinate. One visible relationship identity and one visible solid-line manager are allowed per subordinate; duplicate relationship identities, self-reporting and cycles fail closed, and caller-defined runtime subclasses cannot control identity or temporal comparisons. `build_position_reporting_snapshot(...)` returns deterministic subordinate-to-manager UUID pairs while its routine `repr` redacts those correlation identifiers. This is descriptive organizational evidence only: it neither identifies the worker occupying a seat nor grants employment-decision authority. Persistence and reporting-line mutation remain separate authoritative write-boundary work. + +`PositionSpanOfControlSnapshot` is descriptive organization-design evidence derived only from an exact governed `PositionReportingSnapshot`. It preserves the tenant/business/system coordinate, revalidates the parent graph against forged direct construction, sorts manager Position evidence deterministically, and reports positive direct-report **Position-seat** counts without Person, Employment, or Assignment identifiers. Its canonical evidence is explicitly `structural_workforce_evidence` and `not_authorized_for_employment_decision`. Orgmetra does not encode a universal “good” span: research shows span effects depend on work complexity, hierarchy and organizational goals, so prescriptive interpretation requires separate contextual evidence and accountable human review. `WorkforceCompositionSnapshot` is descriptive reporting evidence, not an employment recommendation or decision. It derives `active` and `leave` workforce composition from the same authoritative bitemporal facts and assignment-integrity rules used by the HRIS kernel. Concurrent employments count one person once for headcount while employment count and staffed FTE retain the actual portfolio shape. Contradictory visible versions, duplicate assignment identities, invalid coverage, and over-allocation fail closed instead of becoming plausible-looking metrics. Canonical snapshot JSON carries only the opaque tenant identifier, report coordinate, aggregate metrics, and schema version; `content_digest()` addresses those exact UTF-8 bytes with SHA-256. @@ -28,4 +31,4 @@ Every historical reconstruction and portfolio/capacity decision requires an expl `AuditOutboxEvent.canonical_json()` is the exact deterministic JSON text that the owning service persists. `AuditOutboxEvent.content_digest()` is SHA-256 over the UTF-8 bytes of that exact text. Callers must not independently serialize `to_cloudevent()` with library defaults and then assume the digest still addresses the stored representation. The Orgmetra PostgreSQL persistence boundary in migration `0003_audit_outbox_persistence.sql` reparses and allowlists that envelope, verifies tenant/event identity and high-impact confirmation, recomputes the digest over the supplied bytes, writes immutable `audit_event_record` evidence, and creates separate `outbox_delivery_record` transport state. The owning service calls `record_audit_outbox_event(...)` inside the same transaction as its business mutation. -This kernel itself does not talk to PostgreSQL, Keyverse, O*NET, or any other service. Persistence, authorization, source retrieval, and UI stay at their adapter boundaries. Production dispatcher claiming, retry scheduling, lease-expiry recovery, retention/export, external delivery receipts, job-analysis persistence, SME workflow, workforce-report authorization/presentation, position-reporting persistence/mutation, and selection-validity computation remain separately proven integration/operability work. \ No newline at end of file +This kernel itself does not talk to PostgreSQL, Keyverse, O*NET, or any other service. Persistence, authorization, source retrieval, and UI stay at their adapter boundaries. Production dispatcher claiming, retry scheduling, lease-expiry recovery, retention/export, external delivery receipts, job-analysis persistence, SME workflow, workforce-report authorization/presentation, position-reporting persistence/mutation, span-of-control persistence/presentation, and selection-validity computation remain separately proven integration/operability work. diff --git a/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py b/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py index b47c9128d..64aae4f38 100644 --- a/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py +++ b/packages/hris-kernel/src/orgmetra_hris_kernel/__init__.py @@ -52,6 +52,11 @@ build_position_reporting_snapshot, ) from orgmetra_hris_kernel.resolution import resolve_bitemporal_facts, resolve_single_valued_fact +from orgmetra_hris_kernel.span_of_control import ( + PositionSpanOfControlError, + PositionSpanOfControlSnapshot, + build_position_span_of_control_snapshot, +) from orgmetra_hris_kernel.workforce import ( WorkforceCompositionSnapshot, build_workforce_composition_snapshot, @@ -80,6 +85,8 @@ "PositionReportingRelationship", "PositionReportingSnapshot", "PositionSeatError", + "PositionSpanOfControlError", + "PositionSpanOfControlSnapshot", "PositionVersion", "RecordedInterval", "SingleValuedFactError", @@ -87,6 +94,7 @@ "TaskKSAOLink", "WorkforceCompositionSnapshot", "build_position_reporting_snapshot", + "build_position_span_of_control_snapshot", "build_workforce_composition_snapshot", "close_recorded_interval", "resolve_bitemporal_facts", diff --git a/packages/hris-kernel/src/orgmetra_hris_kernel/position_reporting.py b/packages/hris-kernel/src/orgmetra_hris_kernel/position_reporting.py index d749d2a6d..c9650b0ba 100644 --- a/packages/hris-kernel/src/orgmetra_hris_kernel/position_reporting.py +++ b/packages/hris-kernel/src/orgmetra_hris_kernel/position_reporting.py @@ -71,10 +71,8 @@ def _freeze_known_at(value: datetime) -> datetime: return wall_time - offset except (OverflowError, ValueError) as exc: raise PositionReportingHierarchyError( - "known_at cannot be represented as a UTC datetime.", - next_action=( - "Use a representable UTC knowledge cutoff, then rebuild the reporting chart." - ), + "known_at UTC instant is outside the supported datetime range.", + next_action="Use a representable authoritative UTC system-knowledge timestamp, then rebuild the chart.", ) from exc @@ -212,8 +210,16 @@ def build_position_reporting_snapshot( ] manager_by_subordinate: dict[UUID, UUID] = {} + seen_relationship_ids: set[UUID] = set() verified_positions: set[UUID] = set() for relationship in visible: + relationship_id = relationship.position_reporting_relationship_id + if relationship_id in seen_relationship_ids: + raise PositionReportingHierarchyError( + "A visible position reporting relationship identity appears more than once.", + next_action="Resolve the duplicate reporting relationship identity, then rebuild the chart.", + ) + seen_relationship_ids.add(relationship_id) subordinate = relationship.subordinate_position_record_id manager = relationship.manager_position_record_id if subordinate in manager_by_subordinate: diff --git a/packages/hris-kernel/src/orgmetra_hris_kernel/span_of_control.py b/packages/hris-kernel/src/orgmetra_hris_kernel/span_of_control.py new file mode 100644 index 000000000..220516e29 --- /dev/null +++ b/packages/hris-kernel/src/orgmetra_hris_kernel/span_of_control.py @@ -0,0 +1,236 @@ +"""PII-minimized structural span-of-control evidence for Position reporting. + +This module counts direct-report Position seats from one governed bitemporal +``PositionReportingSnapshot``. It never counts workers, infers a supervisor from +Person or Assignment, recommends an ideal span, or grants employment-decision +authority. The output is descriptive organization-design evidence only. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import date, datetime, timezone +import hashlib +import json +from uuid import UUID + +from orgmetra_hris_kernel.errors import KernelError +from orgmetra_hris_kernel.position_reporting import PositionReportingSnapshot + +_EVIDENCE_STATE = "structural_workforce_evidence" +_DECISION_AUTHORITY = "not_authorized_for_employment_decision" +_SCHEMA_VERSION = "orgmetra.position_span_of_control.v1" + + +class PositionSpanOfControlError(KernelError): + """Span-of-control evidence is malformed or contradicts its reporting graph.""" + + +def _raise(message: str, next_action: str) -> None: + """Raise one consistent fail-closed structural-evidence error.""" + raise PositionSpanOfControlError(message, next_action=next_action) + + +def _require_uuid(value: UUID, field_name: str) -> None: + """Require one exact non-sentinel opaque UUID.""" + if type(value) is not UUID or value.int in (0, (1 << 128) - 1): + _raise( + f"{field_name} must be an exact non-sentinel UUID.", + "Re-resolve the governed Position reporting evidence, then rebuild the span snapshot.", + ) + + +def _require_business_date(value: date) -> None: + """Require an exact built-in business date.""" + if type(value) is not date: + _raise( + "effective_on must be an exact built-in date.", + "Use the authoritative HR business date, then rebuild the span snapshot.", + ) + + +def _require_utc_knowledge_time(value: datetime) -> None: + """Require the detached built-in UTC instant emitted by the reporting boundary.""" + if type(value) is not datetime or value.tzinfo is not timezone.utc: + _raise( + "known_at must be the exact built-in UTC system-knowledge instant from Position reporting.", + "Rebuild the Position reporting snapshot at the authoritative system-knowledge cutoff.", + ) + + +def _validated_graph( + snapshot: PositionReportingSnapshot, +) -> tuple[tuple[UUID, UUID], ...]: + """Revalidate direct-construction-resistant subordinate-to-manager edges.""" + edges = snapshot.manager_by_subordinate + if type(edges) is not tuple: + _raise( + "Position reporting edge collection must be an exact immutable tuple.", + "Rebuild the Position reporting snapshot through the governed parent boundary.", + ) + + manager_by_subordinate: dict[UUID, UUID] = {} + for edge in edges: + if type(edge) is not tuple or len(edge) != 2: + _raise( + "Each Position reporting edge must be an exact two-item tuple.", + "Rebuild the Position reporting snapshot through the governed parent boundary.", + ) + subordinate, manager = edge + _require_uuid(subordinate, "subordinate_position_record_id") + _require_uuid(manager, "manager_position_record_id") + if subordinate == manager: + _raise( + "A Position cannot report to itself in span-of-control evidence.", + "Correct the reporting relationship, then rebuild the span snapshot.", + ) + if subordinate in manager_by_subordinate: + _raise( + "Position span-of-control evidence contains a duplicate subordinate.", + "Resolve the single visible solid-line manager, then rebuild the span snapshot.", + ) + manager_by_subordinate[subordinate] = manager + + for start in manager_by_subordinate: + seen: set[UUID] = set() + current: UUID | None = start + while current is not None: + if current in seen: + _raise( + "Position span-of-control evidence contains a reporting cycle.", + "Correct one reporting relationship in the cycle, then rebuild the span snapshot.", + ) + seen.add(current) + current = manager_by_subordinate.get(current) + + return tuple(manager_by_subordinate.items()) + + +@dataclass(frozen=True, slots=True, repr=False) +class PositionSpanOfControlSnapshot: + """Deterministic direct-report Position counts at one business/system coordinate.""" + + tenant_record_id: UUID + effective_on: date + known_at: datetime + manager_position_count: int + reporting_position_count: int + span_by_manager: tuple[tuple[UUID, int], ...] + evidence_state: str = field(init=False, default=_EVIDENCE_STATE) + decision_authority: str = field(init=False, default=_DECISION_AUTHORITY) + + def __post_init__(self) -> None: + """Reject contradictory or caller-forged structural output evidence.""" + _require_uuid(self.tenant_record_id, "tenant_record_id") + _require_business_date(self.effective_on) + _require_utc_knowledge_time(self.known_at) + if ( + type(self.manager_position_count) is not int + or self.manager_position_count < 0 + or type(self.reporting_position_count) is not int + or self.reporting_position_count < 0 + ): + _raise( + "Position span-of-control counts must be exact non-negative integers.", + "Rebuild the snapshot from governed Position reporting evidence.", + ) + if type(self.span_by_manager) is not tuple: + _raise( + "span_by_manager must be an exact immutable tuple.", + "Rebuild the snapshot from governed Position reporting evidence.", + ) + + previous_manager_int = -1 + direct_report_total = 0 + for entry in self.span_by_manager: + if type(entry) is not tuple or len(entry) != 2: + _raise( + "span_by_manager entries must be exact two-item tuples.", + "Rebuild the snapshot from governed Position reporting evidence.", + ) + manager, direct_report_count = entry + _require_uuid(manager, "manager_position_record_id") + if type(direct_report_count) is not int or direct_report_count <= 0: + _raise( + "span_by_manager counts must be positive exact integers.", + "Rebuild the snapshot from governed Position reporting evidence.", + ) + if manager.int <= previous_manager_int: + _raise( + "span_by_manager must contain one manager per row in ascending UUID order.", + "Rebuild the snapshot from governed Position reporting evidence.", + ) + previous_manager_int = manager.int + direct_report_total += direct_report_count + + if ( + self.manager_position_count != len(self.span_by_manager) + or self.reporting_position_count != direct_report_total + ): + _raise( + "Position span-of-control counts do not reconcile with span_by_manager.", + "Rebuild the snapshot from one consistent Position reporting graph.", + ) + + def __repr__(self) -> str: + """Keep opaque Position identifiers out of routine logs.""" + return "" + + def canonical_json(self) -> str: + """Return deterministic value-minimized evidence for audit correlation.""" + payload = { + "decision_authority": self.decision_authority, + "effective_on": self.effective_on.isoformat(), + "evidence_state": self.evidence_state, + "known_at": self.known_at.isoformat().replace("+00:00", "Z"), + "manager_position_count": self.manager_position_count, + "reporting_position_count": self.reporting_position_count, + "schema_version": _SCHEMA_VERSION, + "span_by_manager": [ + { + "direct_report_position_count": direct_report_count, + "manager_position_record_id": str(manager), + } + for manager, direct_report_count in self.span_by_manager + ], + "tenant_record_id": str(self.tenant_record_id), + } + return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + def content_digest(self) -> str: + """Return SHA-256 over the exact canonical UTF-8 evidence bytes.""" + return hashlib.sha256(self.canonical_json().encode("utf-8")).hexdigest() + + +def build_position_span_of_control_snapshot( + reporting_snapshot: PositionReportingSnapshot, +) -> PositionSpanOfControlSnapshot: + """Count direct-report Position seats from one governed reporting snapshot. + + The metric is deliberately structural. It counts subordinate Position seats + per manager Position and does not interpret the result as an optimal span, + a worker-performance signal, or a staffing recommendation. + """ + if type(reporting_snapshot) is not PositionReportingSnapshot: + _raise( + "Position span-of-control requires the exact PositionReportingSnapshot runtime type.", + "Build the Position reporting snapshot through the governed HRIS kernel boundary first.", + ) + _require_uuid(reporting_snapshot.tenant_record_id, "tenant_record_id") + _require_business_date(reporting_snapshot.effective_on) + _require_utc_knowledge_time(reporting_snapshot.known_at) + edges = _validated_graph(reporting_snapshot) + + direct_reports_by_manager: dict[UUID, int] = {} + for _subordinate, manager in edges: + direct_reports_by_manager[manager] = direct_reports_by_manager.get(manager, 0) + 1 + + spans = tuple(sorted(direct_reports_by_manager.items(), key=lambda item: item[0].int)) + return PositionSpanOfControlSnapshot( + tenant_record_id=reporting_snapshot.tenant_record_id, + effective_on=reporting_snapshot.effective_on, + known_at=reporting_snapshot.known_at, + manager_position_count=len(spans), + reporting_position_count=len(edges), + span_by_manager=spans, + ) diff --git a/packages/hris-kernel/tests/test_position_reporting.py b/packages/hris-kernel/tests/test_position_reporting.py index 79d51e458..374ed2778 100644 --- a/packages/hris-kernel/tests/test_position_reporting.py +++ b/packages/hris-kernel/tests/test_position_reporting.py @@ -146,6 +146,21 @@ def test_rejects_two_visible_managers_for_one_subordinate() -> None: ) +def test_rejects_duplicate_visible_relationship_identity() -> None: + """One durable reporting relationship identity cannot represent two visible edges.""" + with pytest.raises(PositionReportingHierarchyError, match="identity appears more than once"): + build_position_reporting_snapshot( + [ + relationship(RELATIONSHIP_A, POSITION_B, POSITION_A), + relationship(RELATIONSHIP_A, POSITION_C, POSITION_A), + ], + visible_positions(), + tenant_record_id=TENANT_ALPHA, + effective_on=EFFECTIVE_ON, + known_at=KNOWN_AT, + ) + + def test_rejects_visible_reporting_cycle() -> None: """A position reporting chain must fail closed when it cycles.""" with pytest.raises(PositionReportingHierarchyError, match="form a cycle"): diff --git a/packages/hris-kernel/tests/test_position_reporting_timezone_failure.py b/packages/hris-kernel/tests/test_position_reporting_timezone_failure.py index 972e42b5f..f10defb67 100644 --- a/packages/hris-kernel/tests/test_position_reporting_timezone_failure.py +++ b/packages/hris-kernel/tests/test_position_reporting_timezone_failure.py @@ -39,16 +39,22 @@ def test_timezone_exception_is_normalized_to_governed_error() -> None: ) -def test_unrepresentable_utc_conversion_is_normalized_to_governed_error() -> None: - """UTC normalization overflow cannot escape the reporting boundary.""" - with pytest.raises( - PositionReportingHierarchyError, - match="represented as a UTC datetime", - ): +@pytest.mark.parametrize( + "known_at", + ( + datetime.min.replace(tzinfo=timezone(timedelta(hours=23, minutes=59))), + datetime.max.replace(tzinfo=timezone(timedelta(hours=-14))), + ), +) +def test_timezone_normalization_overflow_is_normalized_to_governed_error( + known_at: datetime, +) -> None: + """An unrepresentable UTC conversion cannot escape as a raw arithmetic error.""" + with pytest.raises(PositionReportingHierarchyError, match="outside the supported datetime range"): build_position_reporting_snapshot( [], [], tenant_record_id=UUID("018f0d35-7b1a-7cc2-8d9c-111111111111"), effective_on=date(2026, 8, 23), - known_at=datetime.max.replace(tzinfo=timezone(timedelta(hours=-14))), + known_at=known_at, ) diff --git a/packages/hris-kernel/tests/test_position_span_of_control.py b/packages/hris-kernel/tests/test_position_span_of_control.py new file mode 100644 index 000000000..3111aca4a --- /dev/null +++ b/packages/hris-kernel/tests/test_position_span_of_control.py @@ -0,0 +1,223 @@ +"""Regression coverage for PII-minimized Position span-of-control evidence.""" + +from datetime import date, datetime, timedelta, timezone +import hashlib +from uuid import UUID + +import pytest + +from orgmetra_hris_kernel.position_reporting import PositionReportingSnapshot +from orgmetra_hris_kernel.span_of_control import ( + PositionSpanOfControlError, + PositionSpanOfControlSnapshot, + build_position_span_of_control_snapshot, +) + +TENANT = UUID("11111111-1111-4111-8111-111111111111") +MANAGER = UUID("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa1") +DIRECTOR = UUID("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa2") +REPORT_A = UUID("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbb1") +REPORT_B = UUID("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbb2") +REPORT_C = UUID("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbb3") +DAY = date(2026, 8, 28) +KNOWN = datetime(2026, 8, 28, 0, 0, tzinfo=timezone.utc) + + +def reporting_snapshot( + edges: tuple[tuple[UUID, UUID], ...], + *, + tenant: UUID = TENANT, + effective_on: date = DAY, + known_at: datetime = KNOWN, +) -> PositionReportingSnapshot: + """Build one direct parent snapshot fixture without adding Person evidence.""" + return PositionReportingSnapshot( + tenant_record_id=tenant, + effective_on=effective_on, + known_at=known_at, + manager_by_subordinate=edges, + ) + + +def test_span_snapshot_counts_direct_reporting_positions_only() -> None: + """The structural metric counts direct-report Position seats, not workers.""" + snapshot = build_position_span_of_control_snapshot( + reporting_snapshot( + ( + (REPORT_A, MANAGER), + (REPORT_B, MANAGER), + (REPORT_C, MANAGER), + (MANAGER, DIRECTOR), + ) + ) + ) + + assert snapshot.manager_position_count == 2 + assert snapshot.reporting_position_count == 4 + assert snapshot.span_by_manager == ((MANAGER, 3), (DIRECTOR, 1)) + assert snapshot.evidence_state == "structural_workforce_evidence" + assert snapshot.decision_authority == "not_authorized_for_employment_decision" + assert "person_record_id" not in snapshot.canonical_json() + assert "assignment_record_id" not in snapshot.canonical_json() + assert snapshot.content_digest() == hashlib.sha256(snapshot.canonical_json().encode("utf-8")).hexdigest() + assert repr(snapshot) == "" + + +def test_empty_reporting_graph_is_valid_zero_span_evidence() -> None: + """An organization with no visible reporting edges yields a deterministic zero snapshot.""" + snapshot = build_position_span_of_control_snapshot(reporting_snapshot(())) + assert snapshot.manager_position_count == 0 + assert snapshot.reporting_position_count == 0 + assert snapshot.span_by_manager == () + + +def test_builder_rejects_non_governed_parent_runtime_type() -> None: + """Caller-defined snapshot-like objects cannot control structural evidence.""" + class SnapshotLike: + pass + + with pytest.raises(PositionSpanOfControlError, match="exact PositionReportingSnapshot"): + build_position_span_of_control_snapshot(SnapshotLike()) # type: ignore[arg-type] + + +@pytest.mark.parametrize( + ("snapshot", "message"), + [ + (reporting_snapshot((), tenant=UUID(int=0)), "tenant_record_id"), + (reporting_snapshot((), effective_on=datetime(2026, 8, 28).date()), "effective_on"), + ( + reporting_snapshot((), known_at=datetime(2026, 8, 28, tzinfo=timezone(timedelta(hours=9)))), + "known_at", + ), + ], +) +def test_builder_rejects_noncanonical_parent_coordinates( + snapshot: PositionReportingSnapshot, + message: str, +) -> None: + """Direct parent construction cannot bypass tenant/business/system coordinate integrity.""" + if message == "effective_on": + class ForgedDate(date): + pass + + object.__setattr__(snapshot, "effective_on", ForgedDate(2026, 8, 28)) + with pytest.raises(PositionSpanOfControlError, match=message): + build_position_span_of_control_snapshot(snapshot) + + +def test_builder_rejects_mutated_edge_container() -> None: + """Reporting edges must retain the immutable tuple representation emitted by the parent boundary.""" + snapshot = reporting_snapshot(((REPORT_A, MANAGER),)) + object.__setattr__(snapshot, "manager_by_subordinate", [(REPORT_A, MANAGER)]) + with pytest.raises(PositionSpanOfControlError, match="edge collection"): + build_position_span_of_control_snapshot(snapshot) + + +def test_builder_rejects_noncanonical_edge_pair() -> None: + """A caller cannot smuggle list-backed reporting pairs into workforce evidence.""" + snapshot = reporting_snapshot(((REPORT_A, MANAGER),)) + object.__setattr__(snapshot, "manager_by_subordinate", ([REPORT_A, MANAGER],)) + with pytest.raises(PositionSpanOfControlError, match="edge must be an exact two-item tuple"): + build_position_span_of_control_snapshot(snapshot) + + +@pytest.mark.parametrize( + ("edges", "message"), + [ + (((REPORT_A, MANAGER), (REPORT_A, DIRECTOR)), "duplicate subordinate"), + (((MANAGER, MANAGER),), "cannot report to itself"), + (((MANAGER, DIRECTOR), (DIRECTOR, MANAGER)), "cycle"), + (((UUID(int=0), MANAGER),), "subordinate_position_record_id"), + (((REPORT_A, UUID(int=(1 << 128) - 1)),), "manager_position_record_id"), + ], +) +def test_builder_fails_closed_on_forged_reporting_graph( + edges: tuple[tuple[UUID, UUID], ...], + message: str, +) -> None: + """Direct construction of a contradictory parent graph cannot forge span evidence.""" + with pytest.raises(PositionSpanOfControlError, match=message): + build_position_span_of_control_snapshot(reporting_snapshot(edges)) + + +def test_direct_output_rejects_inconsistent_counts() -> None: + """Direct output construction cannot claim counts that disagree with the manager spans.""" + with pytest.raises(PositionSpanOfControlError, match="do not reconcile"): + PositionSpanOfControlSnapshot( + tenant_record_id=TENANT, + effective_on=DAY, + known_at=KNOWN, + manager_position_count=1, + reporting_position_count=2, + span_by_manager=((MANAGER, 1),), + ) + + +@pytest.mark.parametrize( + "spans", + [ + ((MANAGER, 0),), + ((MANAGER, True),), + ((DIRECTOR, 1), (MANAGER, 1)), + ((MANAGER, 1), (MANAGER, 1)), + ], +) +def test_direct_output_rejects_noncanonical_span_entries( + spans: tuple[tuple[UUID, int], ...], +) -> None: + """Manager spans remain positive exact integers in UUID order with one row per manager.""" + with pytest.raises(PositionSpanOfControlError, match="span_by_manager"): + PositionSpanOfControlSnapshot( + tenant_record_id=TENANT, + effective_on=DAY, + known_at=KNOWN, + manager_position_count=len(spans), + reporting_position_count=sum(int(value) for _, value in spans), + span_by_manager=spans, + ) + + +@pytest.mark.parametrize( + ("manager_count", "reporting_count"), + [(-1, 0), (True, 0), (0, -1), (0, False)], +) +def test_direct_output_rejects_noncanonical_aggregate_counts( + manager_count: object, + reporting_count: object, +) -> None: + """Aggregate count fields remain exact non-negative integers under direct construction.""" + with pytest.raises(PositionSpanOfControlError, match="exact non-negative integers"): + PositionSpanOfControlSnapshot( + tenant_record_id=TENANT, + effective_on=DAY, + known_at=KNOWN, + manager_position_count=manager_count, # type: ignore[arg-type] + reporting_position_count=reporting_count, # type: ignore[arg-type] + span_by_manager=(), + ) + + +def test_direct_output_rejects_mutable_span_collection() -> None: + """A mutable top-level span collection cannot become canonical audit evidence.""" + with pytest.raises(PositionSpanOfControlError, match="exact immutable tuple"): + PositionSpanOfControlSnapshot( + tenant_record_id=TENANT, + effective_on=DAY, + known_at=KNOWN, + manager_position_count=0, + reporting_position_count=0, + span_by_manager=[], # type: ignore[arg-type] + ) + + +def test_direct_output_rejects_mutable_span_entry() -> None: + """A mutable manager/count entry cannot become canonical audit evidence.""" + with pytest.raises(PositionSpanOfControlError, match="entries must be exact two-item tuples"): + PositionSpanOfControlSnapshot( + tenant_record_id=TENANT, + effective_on=DAY, + known_at=KNOWN, + manager_position_count=1, + reporting_position_count=1, + span_by_manager=([MANAGER, 1],), # type: ignore[arg-type] + )