diff --git a/Neos.ContentGraph.DoctrineDbalAdapter/src/ContentGraphReadModelAdapter.php b/Neos.ContentGraph.DoctrineDbalAdapter/src/ContentGraphReadModelAdapter.php
index bbff12e3e8d..fc9c6f4ddc7 100644
--- a/Neos.ContentGraph.DoctrineDbalAdapter/src/ContentGraphReadModelAdapter.php
+++ b/Neos.ContentGraph.DoctrineDbalAdapter/src/ContentGraphReadModelAdapter.php
@@ -26,7 +26,6 @@
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStream;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreams;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspace;
 use Neos\ContentRepository\Core\SharedModel\Workspace\WorkspaceName;
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspaces;
@@ -104,7 +103,7 @@ public function findContentStreamById(ContentStreamId $contentStreamId): ?Conten
     {
         $contentStreamByIdStatement = <<<SQL
             SELECT
-                id, sourceContentStreamId, status, version, removed
+                id, sourceContentStreamId, version, closed
             FROM
                 {$this->tableNames->contentStream()}
             WHERE
@@ -128,7 +127,7 @@ public function findContentStreams(): ContentStreams
     {
         $contentStreamsStatement = <<<SQL
             SELECT
-                id, sourceContentStreamId, status, version, removed
+                id, sourceContentStreamId, version, closed
             FROM
                 {$this->tableNames->contentStream()}
         SQL;
@@ -160,7 +159,7 @@ private function getBasicWorkspaceQuery(): QueryBuilder
         $queryBuilder = $this->dbal->createQueryBuilder();
 
         return $queryBuilder
-            ->select('ws.name, ws.baseWorkspaceName, ws.currentContentStreamId, cs.sourceContentStreamVersion != scs.version as baseWorkspaceChanged')
+            ->select('ws.name, ws.baseWorkspaceName, ws.currentContentStreamId, cs.sourceContentStreamVersion = scs.version as upToDateWithBase')
             ->from($this->tableNames->workspace(), 'ws')
             ->join('ws', $this->tableNames->contentStream(), 'cs', 'cs.id = ws.currentcontentstreamid')
             ->leftJoin('cs', $this->tableNames->contentStream(), 'scs', 'scs.id = cs.sourceContentStreamId');
@@ -172,15 +171,19 @@ private function getBasicWorkspaceQuery(): QueryBuilder
     private static function workspaceFromDatabaseRow(array $row): Workspace
     {
         $baseWorkspaceName = $row['baseWorkspaceName'] !== null ? WorkspaceName::fromString($row['baseWorkspaceName']) : null;
-        $status = match ($row['baseWorkspaceChanged']) {
+
+        if ($baseWorkspaceName === null) {
             // no base workspace, a root is always up-to-date
-            null => WorkspaceStatus::UP_TO_DATE,
-            // base workspace didnt change (sql 0 is _false_)
-            0 => WorkspaceStatus::UP_TO_DATE,
-            default => WorkspaceStatus::OUTDATED,
-        };
+            $status = WorkspaceStatus::UP_TO_DATE;
+        } elseif ($row['upToDateWithBase'] === 1) {
+            // base workspace didnt change
+            $status = WorkspaceStatus::UP_TO_DATE;
+        } else {
+            // base content stream was removed or contains newer changes
+            $status = WorkspaceStatus::OUTDATED;
+        }
 
-        return new Workspace(
+        return Workspace::create(
             WorkspaceName::fromString($row['name']),
             $baseWorkspaceName,
             ContentStreamId::fromString($row['currentContentStreamId']),
@@ -193,12 +196,11 @@ private static function workspaceFromDatabaseRow(array $row): Workspace
      */
     private static function contentStreamFromDatabaseRow(array $row): ContentStream
     {
-        return new ContentStream(
+        return ContentStream::create(
             ContentStreamId::fromString($row['id']),
             isset($row['sourceContentStreamId']) ? ContentStreamId::fromString($row['sourceContentStreamId']) : null,
-            ContentStreamStatus::from($row['status']),
             Version::fromInteger((int)$row['version']),
-            (bool)$row['removed']
+            (bool)$row['closed'],
         );
     }
 }
diff --git a/Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphProjection.php b/Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphProjection.php
index 27be78643a2..477453656a9 100644
--- a/Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphProjection.php
+++ b/Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphProjection.php
@@ -17,12 +17,11 @@
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Projection\NodeRelationAnchorPoint;
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Repository\DimensionSpacePointsRepository;
 use Neos\ContentGraph\DoctrineDbalAdapter\Domain\Repository\ProjectionContentGraph;
-use Neos\ContentRepository\Core\EventStore\InitiatingEventMetadata;
-use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 use Neos\ContentRepository\Core\DimensionSpace\DimensionSpacePoint;
 use Neos\ContentRepository\Core\DimensionSpace\DimensionSpacePointSet;
 use Neos\ContentRepository\Core\DimensionSpace\OriginDimensionSpacePoint;
 use Neos\ContentRepository\Core\EventStore\EventInterface;
+use Neos\ContentRepository\Core\EventStore\InitiatingEventMetadata;
 use Neos\ContentRepository\Core\Feature\Common\EmbedsContentStreamId;
 use Neos\ContentRepository\Core\Feature\Common\InterdimensionalSiblings;
 use Neos\ContentRepository\Core\Feature\ContentStreamClosing\Event\ContentStreamWasClosed;
@@ -64,15 +63,15 @@
 use Neos\ContentRepository\Core\Infrastructure\DbalSchemaDiff;
 use Neos\ContentRepository\Core\NodeType\NodeTypeName;
 use Neos\ContentRepository\Core\Projection\CheckpointStorageStatusType;
+use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionInterface;
+use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 use Neos\ContentRepository\Core\Projection\ContentGraph\NodeTags;
 use Neos\ContentRepository\Core\Projection\ContentGraph\Timestamps;
 use Neos\ContentRepository\Core\Projection\ProjectionStatus;
-use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphProjectionInterface;
 use Neos\ContentRepository\Core\SharedModel\Node\NodeAggregateClassification;
 use Neos\ContentRepository\Core\SharedModel\Node\NodeAggregateId;
 use Neos\ContentRepository\Core\SharedModel\Node\NodeName;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\EventStore\Model\Event\SequenceNumber;
 use Neos\EventStore\Model\EventEnvelope;
 
@@ -259,12 +258,12 @@ public function inSimulation(\Closure $fn): mixed
 
     private function whenContentStreamWasClosed(ContentStreamWasClosed $event): void
     {
-        $this->updateContentStreamStatus($event->contentStreamId, ContentStreamStatus::CLOSED);
+        $this->closeContentStream($event->contentStreamId);
     }
 
     private function whenContentStreamWasCreated(ContentStreamWasCreated $event): void
     {
-        $this->createContentStream($event->contentStreamId, ContentStreamStatus::CREATED);
+        $this->createContentStream($event->contentStreamId);
     }
 
     private function whenContentStreamWasForked(ContentStreamWasForked $event): void
@@ -303,7 +302,7 @@ private function whenContentStreamWasForked(ContentStreamWasForked $event): void
         // NOTE: as reference edges are attached to Relation Anchor Points (and they are lazily copy-on-written),
         // we do not need to copy reference edges here (but we need to do it during copy on write).
 
-        $this->createContentStream($event->newContentStreamId, ContentStreamStatus::FORKED, $event->sourceContentStreamId, $event->versionOfSourceContentStream);
+        $this->createContentStream($event->newContentStreamId, $event->sourceContentStreamId, $event->versionOfSourceContentStream);
     }
 
     private function whenContentStreamWasRemoved(ContentStreamWasRemoved $event): void
@@ -353,7 +352,7 @@ private function whenContentStreamWasRemoved(ContentStreamWasRemoved $event): vo
 
     private function whenContentStreamWasReopened(ContentStreamWasReopened $event): void
     {
-        $this->updateContentStreamStatus($event->contentStreamId, $event->previousState);
+        $this->reopenContentStream($event->contentStreamId);
     }
 
     private function whenDimensionShineThroughWasAdded(DimensionShineThroughWasAdded $event): void
@@ -708,9 +707,6 @@ private function whenRootNodeAggregateWithNodeWasCreated(RootNodeAggregateWithNo
     private function whenRootWorkspaceWasCreated(RootWorkspaceWasCreated $event): void
     {
         $this->createWorkspace($event->workspaceName, null, $event->newContentStreamId);
-
-        // the content stream is in use now
-        $this->updateContentStreamStatus($event->newContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
     }
 
     private function whenSubtreeWasTagged(SubtreeWasTagged $event): void
@@ -730,69 +726,41 @@ private function whenWorkspaceBaseWorkspaceWasChanged(WorkspaceBaseWorkspaceWasC
 
     private function whenWorkspaceRebaseFailed(WorkspaceRebaseFailed $event): void
     {
-        $this->updateContentStreamStatus($event->candidateContentStreamId, ContentStreamStatus::REBASE_ERROR);
+        // legacy handling:
+        // before https://github.com/neos/neos-development-collection/pull/4965 this event was emitted and set the content stream status to `REBASE_ERROR`
+        // instead of setting the error state on replay for old events we make it almost behave like if the rebase had failed today: reopen the workspaces content stream id
+        // the candidateContentStreamId will be removed by the ContentStreamPruner
+        $this->reopenContentStream($event->sourceContentStreamId);
     }
 
     private function whenWorkspaceWasCreated(WorkspaceWasCreated $event): void
     {
         $this->createWorkspace($event->workspaceName, $event->baseWorkspaceName, $event->newContentStreamId);
-
-        // the content stream is in use now
-        $this->updateContentStreamStatus($event->newContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
     }
 
     private function whenWorkspaceWasDiscarded(WorkspaceWasDiscarded $event): void
     {
         $this->updateWorkspaceContentStreamId($event->workspaceName, $event->newContentStreamId);
-
-        // the new content stream is in use now
-        $this->updateContentStreamStatus($event->newContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
-        // the previous content stream is no longer in use
-        $this->updateContentStreamStatus($event->previousContentStreamId, ContentStreamStatus::NO_LONGER_IN_USE);
     }
 
     private function whenWorkspaceWasPartiallyDiscarded(WorkspaceWasPartiallyDiscarded $event): void
     {
         $this->updateWorkspaceContentStreamId($event->workspaceName, $event->newContentStreamId);
-
-        // the new content stream is in use now
-        $this->updateContentStreamStatus($event->newContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
-
-        // the previous content stream is no longer in use
-        $this->updateContentStreamStatus($event->previousContentStreamId, ContentStreamStatus::NO_LONGER_IN_USE);
     }
 
     private function whenWorkspaceWasPartiallyPublished(WorkspaceWasPartiallyPublished $event): void
     {
         $this->updateWorkspaceContentStreamId($event->sourceWorkspaceName, $event->newSourceContentStreamId);
-
-        // the new content stream is in use now
-        $this->updateContentStreamStatus($event->newSourceContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
-
-        // the previous content stream is no longer in use
-        $this->updateContentStreamStatus($event->previousSourceContentStreamId, ContentStreamStatus::NO_LONGER_IN_USE);
     }
 
     private function whenWorkspaceWasPublished(WorkspaceWasPublished $event): void
     {
         $this->updateWorkspaceContentStreamId($event->sourceWorkspaceName, $event->newSourceContentStreamId);
-
-        // the new content stream is in use now
-        $this->updateContentStreamStatus($event->newSourceContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
-
-        // the previous content stream is no longer in use
-        $this->updateContentStreamStatus($event->previousSourceContentStreamId, ContentStreamStatus::NO_LONGER_IN_USE);
     }
 
     private function whenWorkspaceWasRebased(WorkspaceWasRebased $event): void
     {
         $this->updateWorkspaceContentStreamId($event->workspaceName, $event->newContentStreamId);
-
-        // the new content stream is in use now
-        $this->updateContentStreamStatus($event->newContentStreamId, ContentStreamStatus::IN_USE_BY_WORKSPACE);
-
-        // the previous content stream is no longer in use
-        $this->updateContentStreamStatus($event->previousContentStreamId, ContentStreamStatus::NO_LONGER_IN_USE);
     }
 
     private function whenWorkspaceWasRemoved(WorkspaceWasRemoved $event): void
diff --git a/Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphSchemaBuilder.php b/Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphSchemaBuilder.php
index 0ce0ee51e2f..0d13ff5ef91 100644
--- a/Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphSchemaBuilder.php
+++ b/Neos.ContentGraph.DoctrineDbalAdapter/src/DoctrineDbalContentGraphSchemaBuilder.php
@@ -127,9 +127,7 @@ private function createContentStreamTable(): Table
             (new Column('version', Type::getType(Types::INTEGER)))->setNotnull(true),
             DbalSchemaFactory::columnForContentStreamId('sourceContentStreamId')->setNotnull(false),
             (new Column('sourceContentStreamVersion', Type::getType(Types::INTEGER)))->setNotnull(false),
-            // Should become a DB ENUM (unclear how to configure with DBAL) or int (latter needs adaption to code)
-            (new Column('status', Type::getType(Types::BINARY)))->setLength(20)->setNotnull(true),
-            (new Column('removed', Type::getType(Types::BOOLEAN)))->setDefault(false)->setNotnull(false),
+            (new Column('closed', Type::getType(Types::BOOLEAN)))->setDefault(false)->setNotnull(true),
         ]);
     }
 
diff --git a/Neos.ContentGraph.DoctrineDbalAdapter/src/Domain/Projection/Feature/ContentStream.php b/Neos.ContentGraph.DoctrineDbalAdapter/src/Domain/Projection/Feature/ContentStream.php
index 10a4392aca1..50cd71bb02e 100644
--- a/Neos.ContentGraph.DoctrineDbalAdapter/src/Domain/Projection/Feature/ContentStream.php
+++ b/Neos.ContentGraph.DoctrineDbalAdapter/src/Domain/Projection/Feature/ContentStream.php
@@ -5,7 +5,6 @@
 namespace Neos\ContentGraph\DoctrineDbalAdapter\Domain\Projection\Feature;
 
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\EventStore\Model\Event\Version;
 
 /**
@@ -15,35 +14,41 @@
  */
 trait ContentStream
 {
-    private function createContentStream(ContentStreamId $contentStreamId, ContentStreamStatus $status, ?ContentStreamId $sourceContentStreamId = null, ?Version $sourceVersion = null): void
+    private function createContentStream(ContentStreamId $contentStreamId, ?ContentStreamId $sourceContentStreamId = null, ?Version $sourceVersion = null): void
     {
         $this->dbal->insert($this->tableNames->contentStream(), [
             'id' => $contentStreamId->value,
             'version' => 0,
             'sourceContentStreamId' => $sourceContentStreamId?->value,
-            'sourceContentStreamVersion' => $sourceVersion?->value,
-            'status' => $status->value,
+            'sourceContentStreamVersion' => $sourceVersion?->value
         ]);
     }
 
-    private function updateContentStreamStatus(ContentStreamId $contentStreamId, ContentStreamStatus $status): void
+    private function closeContentStream(ContentStreamId $contentStreamId): void
     {
         $this->dbal->update($this->tableNames->contentStream(), [
-            'status' => $status->value,
+            'closed' => 1,
         ], [
             'id' => $contentStreamId->value
         ]);
     }
 
-    private function removeContentStream(ContentStreamId $contentStreamId): void
+    private function reopenContentStream(ContentStreamId $contentStreamId): void
     {
         $this->dbal->update($this->tableNames->contentStream(), [
-            'removed' => true,
+            'closed' => 0,
         ], [
             'id' => $contentStreamId->value
         ]);
     }
 
+    private function removeContentStream(ContentStreamId $contentStreamId): void
+    {
+        $this->dbal->delete($this->tableNames->contentStream(), [
+            'id' => $contentStreamId->value
+        ]);
+    }
+
     private function updateContentStreamVersion(ContentStreamId $contentStreamId, Version $version): void
     {
         $this->dbal->update($this->tableNames->contentStream(), [
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/01-RootNodeCreation/01-CreateRootNodeAggregateWithNode_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/01-RootNodeCreation/01-CreateRootNodeAggregateWithNode_ConstraintChecks.feature
index be7df0bd441..10960530bf2 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/01-RootNodeCreation/01-CreateRootNodeAggregateWithNode_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/01-RootNodeCreation/01-CreateRootNodeAggregateWithNode_ConstraintChecks.feature
@@ -39,7 +39,7 @@ Feature: Create a root node aggregate
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to create a root node aggregate in a closed content stream:
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     And the command CreateRootNodeAggregateWithNode is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/02-NodeCreation/01-CreateNodeAggregateWithNode_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/02-NodeCreation/01-CreateNodeAggregateWithNode_ConstraintChecks.feature
index 6e608301510..19a5500ab86 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/02-NodeCreation/01-CreateNodeAggregateWithNode_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/02-NodeCreation/01-CreateNodeAggregateWithNode_ConstraintChecks.feature
@@ -53,7 +53,7 @@ Feature: Create node aggregate with node
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to create a node aggregate in a workspace whose content stream is closed:
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     And the command CreateNodeAggregateWithNode is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/03-NodeVariation/01-CreateNodeVariant_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/03-NodeVariation/01-CreateNodeVariant_ConstraintChecks.feature
index cbe2098d6da..5df29d7fe45 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/03-NodeVariation/01-CreateNodeVariant_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/03-NodeVariation/01-CreateNodeVariant_ConstraintChecks.feature
@@ -47,7 +47,7 @@ Feature: Create node variant
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to create a variant in a workspace that does not exist
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     And the command CreateNodeVariant is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/04-NodeModification/01-SetNodeProperties_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/04-NodeModification/01-SetNodeProperties_ConstraintChecks.feature
index 3dd1295af19..968798de225 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/04-NodeModification/01-SetNodeProperties_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/04-NodeModification/01-SetNodeProperties_ConstraintChecks.feature
@@ -44,7 +44,7 @@ Feature: Set node properties: Constraint checks
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to set properties in a workspace whose content stream is closed
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command SetNodeProperties is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/05-NodeReferencing/01-SetNodeReferences_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/05-NodeReferencing/01-SetNodeReferences_ConstraintChecks.feature
index 7f6aa1c37ca..7dc5687b611 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/05-NodeReferencing/01-SetNodeReferences_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/05-NodeReferencing/01-SetNodeReferences_ConstraintChecks.feature
@@ -55,7 +55,7 @@ Feature: Constraint checks on SetNodeReferences
       | berta-destinode   | Neos.ContentRepository.Testing:ReferencedNode     | lady-eleonode-rootford |
 
   Scenario: Try to reference nodes in a workspace whose content stream is closed
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command SetNodeReferences is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/06-NodeDisabling/01-DisableNodeAggregate_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/06-NodeDisabling/01-DisableNodeAggregate_ConstraintChecks.feature
index 4a353636c61..91c845e2b27 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/06-NodeDisabling/01-DisableNodeAggregate_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/06-NodeDisabling/01-DisableNodeAggregate_ConstraintChecks.feature
@@ -38,7 +38,7 @@ Feature: Constraint checks on node aggregate disabling
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to disable a node aggregate in a workspace whose content stream is closed
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command DisableNodeAggregate is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/07-NodeRemoval/01-RemoveNodeAggregate_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/07-NodeRemoval/01-RemoveNodeAggregate_ConstraintChecks.feature
index c11417b9ad0..e7e35cb4a5a 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/07-NodeRemoval/01-RemoveNodeAggregate_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/07-NodeRemoval/01-RemoveNodeAggregate_ConstraintChecks.feature
@@ -43,7 +43,7 @@ Feature: Remove NodeAggregate
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to remove a node aggregate in a workspace whose content stream is closed
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command RemoveNodeAggregate is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/08-NodeMove/01-MoveNodes_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/08-NodeMove/01-MoveNodes_ConstraintChecks.feature
index 3ebce961613..37f9a1c8523 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/08-NodeMove/01-MoveNodes_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/08-NodeMove/01-MoveNodes_ConstraintChecks.feature
@@ -59,7 +59,7 @@ Feature: Move node to a new parent / within the current parent before a sibling
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to move a node in a workspace whose content stream is closed:
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command MoveNodeAggregate is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/09-NodeRenaming/01-ChangeNodeAggregateName_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/09-NodeRenaming/01-ChangeNodeAggregateName_ConstraintChecks.feature
index 50c499912b4..b8a003c26a4 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/09-NodeRenaming/01-ChangeNodeAggregateName_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/09-NodeRenaming/01-ChangeNodeAggregateName_ConstraintChecks.feature
@@ -43,7 +43,7 @@ Feature: Change node name
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to rename a node aggregate in a workspace whose content stream is closed:
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command ChangeNodeAggregateName is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/11-NodeTypeChange/01-ChangeNodeAggregateType_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/11-NodeTypeChange/01-ChangeNodeAggregateType_ConstraintChecks.feature
index 55148afde37..9ee1fafd150 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/11-NodeTypeChange/01-ChangeNodeAggregateType_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/11-NodeTypeChange/01-ChangeNodeAggregateType_ConstraintChecks.feature
@@ -78,7 +78,7 @@ Feature: Change node aggregate type - basic error cases
     Then the last command should have thrown an exception of type "WorkspaceDoesNotExist"
 
   Scenario: Try to change the node aggregate type in a workspace whose content stream is closed
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     And the command ChangeNodeAggregateType is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/ContentStreamClosing/01-CloseContentStream_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/ContentStreamClosing/01-CloseContentStream_ConstraintChecks.feature
deleted file mode 100644
index 595c2e53055..00000000000
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/ContentStreamClosing/01-CloseContentStream_ConstraintChecks.feature
+++ /dev/null
@@ -1,30 +0,0 @@
-@contentrepository @adapters=DoctrineDBAL
-Feature: Constraint check test cases for closing content streams
-
-  Background:
-    Given using no content dimensions
-    And using the following node types:
-    """yaml
-    Neos.ContentRepository:Root: {}
-    """
-    And using identifier "default", I define a content repository
-    And I am in content repository "default"
-    And the command CreateRootWorkspace is executed with payload:
-      | Key                | Value           |
-      | workspaceName      | "live"          |
-      | newContentStreamId | "cs-identifier" |
-
-  Scenario: Try to close a non-existing content stream:
-    And the command CloseContentStream is executed with payload and exceptions are caught:
-      | Key             | Value            |
-      | contentStreamId | "i-do-not-exist" |
-    Then the last command should have thrown an exception of type "ContentStreamDoesNotExistYet"
-
-  Scenario: Try to close a content stream that is already closed:
-    When the command CloseContentStream is executed with payload:
-      | Key             | Value           |
-      | contentStreamId | "cs-identifier" |
-    And the command CloseContentStream is executed with payload and exceptions are caught:
-      | Key             | Value           |
-      | contentStreamId | "cs-identifier" |
-    Then the last command should have thrown an exception of type "ContentStreamIsClosed"
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/ContentStreamForking/01-ForkContentStream_ConstraintChecks.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/ContentStreamForking/01-ForkContentStream_ConstraintChecks.feature
index 5b3a1fd7723..2aa54f9856f 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/ContentStreamForking/01-ForkContentStream_ConstraintChecks.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/ContentStreamForking/01-ForkContentStream_ConstraintChecks.feature
@@ -50,7 +50,7 @@ Feature: ForkContentStream Without Dimensions
       | propertiesToUnset            | {}                                                      |
 
   Scenario: Try to create a workspace with the base workspace referring to a closed content stream
-    When the command CloseContentStream is executed with payload:
+    When the event ContentStreamWasClosed was published with payload:
       | Key             | Value           |
       | contentStreamId | "cs-identifier" |
     When the command CreateWorkspace is executed with payload and exceptions are caught:
diff --git a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/Workspaces/PruneContentStreams.feature b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/Workspaces/PruneContentStreams.feature
index dadd9e7e112..c64ec997a14 100644
--- a/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/Workspaces/PruneContentStreams.feature
+++ b/Neos.ContentRepository.BehavioralTests/Tests/Behavior/Features/Workspaces/PruneContentStreams.feature
@@ -19,36 +19,38 @@ Feature: If content streams are not in use anymore by the workspace, they can be
       | nodeAggregateId | "root-node"                   |
       | nodeTypeName    | "Neos.ContentRepository:Root" |
 
-  Scenario: content streams are marked as IN_USE_BY_WORKSPACE properly after creation
-    Then the content stream "cs-identifier" has status "IN_USE_BY_WORKSPACE"
+  #
+  # Before Neos 9 beta 15 (publishing version 3 #5301), dangling content streams were not removed during publishing, discard or rebase
+  # The first scenarios assert that the automatic deletion works correctly
+  #
+
+  Scenario: content streams are in use after creation
     Then I expect the content stream "non-existing" to not exist
+    Then I expect the content stream "cs-identifier" to exist
 
-  Scenario: on creating a nested workspace, the new content stream is marked as IN_USE_BY_WORKSPACE.
-    When the command CreateWorkspace is executed with payload:
-      | Key                | Value                |
-      | workspaceName      | "user-test"          |
-      | baseWorkspaceName  | "live"               |
-      | newContentStreamId | "user-cs-identifier" |
+    Then I expect the content stream pruner status output:
+    """
+    Okay. No dangling streams found
 
-    Then the content stream "user-cs-identifier" has status "IN_USE_BY_WORKSPACE"
+    Okay. No pruneable streams in the event stream
+    """
 
-  Scenario: when rebasing a nested workspace, the new content stream will be marked as IN_USE_BY_WORKSPACE; and the old content stream is NO_LONGER_IN_USE.
+  Scenario: on creating a nested workspace, the new content stream is not pruned
     When the command CreateWorkspace is executed with payload:
       | Key                | Value                |
       | workspaceName      | "user-test"          |
       | baseWorkspaceName  | "live"               |
       | newContentStreamId | "user-cs-identifier" |
-    When the command RebaseWorkspace is executed with payload:
-      | Key           | Value       |
-      | workspaceName | "user-test" |
-      | rebaseErrorHandlingStrategy | "force"               |
+    Then I expect the content stream "user-cs-identifier" to exist
 
-    When I am in workspace "user-test" and dimension space point {}
-    Then the current content stream has status "IN_USE_BY_WORKSPACE"
-    And the content stream "user-cs-identifier" has status "NO_LONGER_IN_USE"
+    Then I expect the content stream pruner status output:
+    """
+    Okay. No dangling streams found
 
+    Okay. No pruneable streams in the event stream
+    """
 
-  Scenario: when pruning content streams, NO_LONGER_IN_USE content streams will be properly cleaned from the graph projection.
+  Scenario: no longer in use content streams will be properly cleaned from the graph projection.
     When the command CreateWorkspace is executed with payload:
       | Key                | Value                |
       | workspaceName      | "user-test"          |
@@ -65,13 +67,13 @@ Feature: If content streams are not in use anymore by the workspace, they can be
       | rebaseErrorHandlingStrategy | "force"               |
     # now, we have one unused content stream (the old content stream of the user-test workspace)
 
-    When I prune unused content streams
     Then I expect the content stream "user-cs-identifier" to not exist
 
     When I am in workspace "user-test" and dimension space point {}
+    # todo test that the graph projection really is cleaned up and that no hierarchy stil exist?
     Then I expect node aggregate identifier "root-node" to lead to node user-cs-identifier-rebased;root-node;{}
 
-  Scenario: NO_LONGER_IN_USE content streams can be cleaned up completely (simple case)
+  Scenario: no longer in use content streams can be cleaned up completely (simple case)
 
     When the command CreateWorkspace is executed with payload:
       | Key                | Value                |
@@ -79,19 +81,37 @@ Feature: If content streams are not in use anymore by the workspace, they can be
       | baseWorkspaceName  | "live"               |
       | newContentStreamId | "user-cs-identifier" |
     When the command RebaseWorkspace is executed with payload:
-      | Key           | Value       |
-      | workspaceName | "user-test" |
+      | Key                    | Value                        |
+      | workspaceName          | "user-test"                  |
+      | rebasedContentStreamId | "user-cs-identifier-rebased" |
       | rebaseErrorHandlingStrategy | "force"               |
 
+    Then I expect the content stream "user-cs-identifier-rebased" to exist
+    Then I expect the content stream "user-cs-identifier" to not exist
+
     # now, we have one unused content stream (the old content stream of the user-test workspace)
 
-    When I prune unused content streams
+    Then I expect the content stream pruner status output:
+    """
+    Okay. No dangling streams found
+
+    Removed content streams that can be pruned from the event stream
+      id: user-cs-identifier previous state: no longer in use
+    To prune the removed streams from the event stream run ./flow contentStream:pruneRemovedFromEventstream
+    """
+
     And I prune removed content streams from the event stream
 
     Then I expect exactly 0 events to be published on stream "ContentStream:user-cs-identifier"
 
+    Then I expect the content stream pruner status output:
+    """
+    Okay. No dangling streams found
+
+    Okay. No pruneable streams in the event stream
+    """
 
-  Scenario: NO_LONGER_IN_USE content streams are only cleaned up if no other content stream which is still in use depends on it
+  Scenario: no longer in use content streams are only cleaned up if no other content stream which is still in use depends on it
     # we build a "review" workspace, and then a "user-test" workspace depending on the review workspace.
     When the command CreateWorkspace is executed with payload:
       | Key                | Value                  |
@@ -104,7 +124,7 @@ Feature: If content streams are not in use anymore by the workspace, they can be
       | baseWorkspaceName  | "review"             |
       | newContentStreamId | "user-cs-identifier" |
 
-    # now, we rebase the "review" workspace, effectively marking the "review-cs-identifier" content stream as NO_LONGER_IN_USE.
+    # now, we rebase the "review" workspace, effectively marking the "review-cs-identifier" content stream as no longer in use.
     # however, we are not allowed to drop the content stream from the event store yet, because the "user-cs-identifier" is based
     # on the (no-longer-in-direct-use) review-cs-identifier.
     When the command RebaseWorkspace is executed with payload:
@@ -112,8 +132,14 @@ Feature: If content streams are not in use anymore by the workspace, they can be
       | workspaceName | "review" |
       | rebaseErrorHandlingStrategy | "force"               |
 
-    When I prune unused content streams
     And I prune removed content streams from the event stream
 
     # the events should still exist
     Then I expect exactly 3 events to be published on stream "ContentStream:review-cs-identifier"
+
+    Then I expect the content stream pruner status output:
+    """
+    Okay. No dangling streams found
+
+    Okay. No pruneable streams in the event stream
+    """
diff --git a/Neos.ContentRepository.Core/Classes/CommandHandler/CommandHandlingDependencies.php b/Neos.ContentRepository.Core/Classes/CommandHandler/CommandHandlingDependencies.php
index 09b97dc9293..9e726cacea1 100644
--- a/Neos.ContentRepository.Core/Classes/CommandHandler/CommandHandlingDependencies.php
+++ b/Neos.ContentRepository.Core/Classes/CommandHandler/CommandHandlingDependencies.php
@@ -18,7 +18,6 @@
 use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 use Neos\ContentRepository\Core\SharedModel\Exception\WorkspaceDoesNotExist;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspace;
 use Neos\ContentRepository\Core\SharedModel\Workspace\WorkspaceName;
 use Neos\EventStore\Model\Event\Version;
@@ -46,17 +45,16 @@ public function getContentStreamVersion(ContentStreamId $contentStreamId): Versi
 
     public function contentStreamExists(ContentStreamId $contentStreamId): bool
     {
-        $cs = $this->contentGraphReadModel->findContentStreamById($contentStreamId);
-        return $cs !== null && !$cs->removed;
+        return $this->contentGraphReadModel->findContentStreamById($contentStreamId) !== null;
     }
 
-    public function getContentStreamStatus(ContentStreamId $contentStreamId): ContentStreamStatus
+    public function isContentStreamClosed(ContentStreamId $contentStreamId): bool
     {
         $contentStream = $this->contentGraphReadModel->findContentStreamById($contentStreamId);
         if ($contentStream === null) {
-            throw new \InvalidArgumentException(sprintf('Failed to find content stream with id "%s"', $contentStreamId->value), 1716902219);
+            throw new \InvalidArgumentException(sprintf('Failed to find content stream with id "%s"', $contentStreamId->value), 1729863973);
         }
-        return $contentStream->status;
+        return $contentStream->isClosed;
     }
 
     public function findWorkspaceByName(WorkspaceName $workspaceName): ?Workspace
diff --git a/Neos.ContentRepository.Core/Classes/Factory/ContentRepositoryFactory.php b/Neos.ContentRepository.Core/Classes/Factory/ContentRepositoryFactory.php
index f8244afea53..d424f39138e 100644
--- a/Neos.ContentRepository.Core/Classes/Factory/ContentRepositoryFactory.php
+++ b/Neos.ContentRepository.Core/Classes/Factory/ContentRepositoryFactory.php
@@ -23,7 +23,6 @@
 use Neos\ContentRepository\Core\DimensionSpace\InterDimensionalVariationGraph;
 use Neos\ContentRepository\Core\EventStore\EventNormalizer;
 use Neos\ContentRepository\Core\EventStore\EventPersister;
-use Neos\ContentRepository\Core\Feature\ContentStreamCommandHandler;
 use Neos\ContentRepository\Core\Feature\DimensionSpaceAdjustment\DimensionSpaceCommandHandler;
 use Neos\ContentRepository\Core\Feature\NodeAggregateCommandHandler;
 use Neos\ContentRepository\Core\Feature\NodeDuplication\NodeDuplicationCommandHandler;
@@ -121,7 +120,6 @@ public function getOrBuild(): ContentRepository
         );
 
         $publicCommandBus = $commandBusForRebaseableCommands->withAdditionalHandlers(
-            new ContentStreamCommandHandler(),
             new WorkspaceCommandHandler(
                 $commandSimulatorFactory,
                 $this->projectionFactoryDependencies->eventStore,
diff --git a/Neos.ContentRepository.Core/Classes/Feature/Common/ConstraintChecks.php b/Neos.ContentRepository.Core/Classes/Feature/Common/ConstraintChecks.php
index 30b67c1de78..7957ab4de30 100644
--- a/Neos.ContentRepository.Core/Classes/Feature/Common/ConstraintChecks.php
+++ b/Neos.ContentRepository.Core/Classes/Feature/Common/ConstraintChecks.php
@@ -61,7 +61,6 @@
 use Neos\ContentRepository\Core\SharedModel\Node\PropertyName;
 use Neos\ContentRepository\Core\SharedModel\Node\ReferenceName;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\ContentRepository\Core\SharedModel\Workspace\WorkspaceName;
 use Neos\EventStore\Model\EventStream\ExpectedVersion;
 
@@ -88,8 +87,8 @@ protected function requireContentStream(
                 1521386692
             );
         }
-        $state = $commandHandlingDependencies->getContentStreamStatus($contentStreamId);
-        if ($state === ContentStreamStatus::CLOSED) {
+
+        if ($commandHandlingDependencies->isContentStreamClosed($contentStreamId)) {
             throw new ContentStreamIsClosed(
                 'Content stream "' . $contentStreamId->value . '" is closed.',
                 1710260081
diff --git a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamClosing/Command/CloseContentStream.php b/Neos.ContentRepository.Core/Classes/Feature/ContentStreamClosing/Command/CloseContentStream.php
deleted file mode 100644
index 3234b633586..00000000000
--- a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamClosing/Command/CloseContentStream.php
+++ /dev/null
@@ -1,52 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Neos\ContentRepository\Core\Feature\ContentStreamClosing\Command;
-
-/*
- * This file is part of the Neos.ContentRepository package.
- *
- * (c) Contributors of the Neos Project - www.neos.io
- *
- * This package is Open Source Software. For the full copyright and license
- * information, please view the LICENSE file which was distributed with this
- * source code.
- */
-
-use Neos\ContentRepository\Core\CommandHandler\CommandInterface;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-
-/**
- * @internal only exposed for testing purposes. You must not use this command directly.
- * Direct use will leave your content stream in a closed state.
- */
-final readonly class CloseContentStream implements CommandInterface
-{
-    /**
-     * @param ContentStreamId $contentStreamId The id of the content stream to close
-     */
-    private function __construct(
-        public ContentStreamId $contentStreamId,
-    ) {
-    }
-
-    /**
-     * @param ContentStreamId $contentStreamId The id of the content stream to close
-     */
-    public static function create(ContentStreamId $contentStreamId): self
-    {
-        return new self($contentStreamId);
-    }
-
-    /**
-     * @param array<string,string> $array
-     * @internal only used for testcases
-     */
-    public static function fromArray(array $array): self
-    {
-        return new self(
-            ContentStreamId::fromString($array['contentStreamId']),
-        );
-    }
-}
diff --git a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamClosing/Command/ReopenContentStream.php b/Neos.ContentRepository.Core/Classes/Feature/ContentStreamClosing/Command/ReopenContentStream.php
deleted file mode 100644
index 10280ae9890..00000000000
--- a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamClosing/Command/ReopenContentStream.php
+++ /dev/null
@@ -1,63 +0,0 @@
-<?php
-
-/*
- * This file is part of the Neos.ContentRepository.Core  package.
- *
- * (c) Contributors of the Neos Project - www.neos.io
- *
- * This package is Open Source Software. For the full copyright and license
- * information, please view the LICENSE file which was distributed with this
- * source code.
- */
-
-declare(strict_types=1);
-
-namespace Neos\ContentRepository\Core\Feature\ContentStreamClosing\Command;
-
-use Neos\ContentRepository\Core\CommandHandler\CommandInterface;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
-
-/**
- * @internal implementation detail. You must not use this command directly.
- * Direct use may lead to hard to revert senseless state in your content repository.
- * Please use the higher level workspace commands instead.
- */
-final readonly class ReopenContentStream implements CommandInterface
-{
-    /**
-     * @param ContentStreamId $contentStreamId The id of the content stream to reopen
-     * @param ContentStreamStatus $previousState The state the content stream was in before closing and is to be reset to
-     */
-    private function __construct(
-        public ContentStreamId $contentStreamId,
-        public ContentStreamStatus $previousState
-    ) {
-    }
-
-    /**
-     * @param ContentStreamId $contentStreamId The id of the content stream to reopen
-     * @param ContentStreamStatus $previousState The state the content stream was in before closing and is to be reset to
-     */
-    public static function create(
-        ContentStreamId $contentStreamId,
-        ContentStreamStatus $previousState
-    ): self {
-        return new self(
-            $contentStreamId,
-            $previousState
-        );
-    }
-
-    /**
-     * @param array<string,string> $array
-     * @internal only used for testcases
-     */
-    public static function fromArray(array $array): self
-    {
-        return new self(
-            ContentStreamId::fromString($array['contentStreamId']),
-            ContentStreamStatus::from($array['previousState']),
-        );
-    }
-}
diff --git a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamClosing/Event/ContentStreamWasReopened.php b/Neos.ContentRepository.Core/Classes/Feature/ContentStreamClosing/Event/ContentStreamWasReopened.php
index 4b0248803f7..b783ee99fb5 100644
--- a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamClosing/Event/ContentStreamWasReopened.php
+++ b/Neos.ContentRepository.Core/Classes/Feature/ContentStreamClosing/Event/ContentStreamWasReopened.php
@@ -17,7 +17,6 @@
 use Neos\ContentRepository\Core\EventStore\EventInterface;
 use Neos\ContentRepository\Core\Feature\Common\EmbedsContentStreamId;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 
 /**
  * @api events are the persistence-API of the content repository
@@ -25,8 +24,7 @@
 final readonly class ContentStreamWasReopened implements EventInterface, EmbedsContentStreamId
 {
     public function __construct(
-        public ContentStreamId $contentStreamId,
-        public ContentStreamStatus $previousState,
+        public ContentStreamId $contentStreamId
     ) {
     }
 
@@ -38,16 +36,14 @@ public function getContentStreamId(): ContentStreamId
     public static function fromArray(array $values): self
     {
         return new self(
-            ContentStreamId::fromString($values['contentStreamId']),
-            ContentStreamStatus::from($values['previousState']),
+            ContentStreamId::fromString($values['contentStreamId'])
         );
     }
 
     public function jsonSerialize(): array
     {
         return [
-            'contentStreamId' => $this->contentStreamId,
-            'previousState' => $this->previousState,
+            'contentStreamId' => $this->contentStreamId
         ];
     }
 }
diff --git a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamCommandHandler.php b/Neos.ContentRepository.Core/Classes/Feature/ContentStreamCommandHandler.php
deleted file mode 100644
index 1b8400833d1..00000000000
--- a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamCommandHandler.php
+++ /dev/null
@@ -1,72 +0,0 @@
-<?php
-
-/*
- * This file is part of the Neos.ContentRepository package.
- *
- * (c) Contributors of the Neos Project - www.neos.io
- *
- * This package is Open Source Software. For the full copyright and license
- * information, please view the LICENSE file which was distributed with this
- * source code.
- */
-
-declare(strict_types=1);
-
-namespace Neos\ContentRepository\Core\Feature;
-
-use Neos\ContentRepository\Core\CommandHandler\CommandHandlerInterface;
-use Neos\ContentRepository\Core\CommandHandler\CommandInterface;
-use Neos\ContentRepository\Core\CommandHandler\CommandHandlingDependencies;
-use Neos\ContentRepository\Core\ContentRepository;
-use Neos\ContentRepository\Core\EventStore\EventsToPublish;
-use Neos\ContentRepository\Core\Feature\ContentStreamClosing\Command\CloseContentStream;
-use Neos\ContentRepository\Core\Feature\ContentStreamClosing\Command\ReopenContentStream;
-use Neos\ContentRepository\Core\Feature\ContentStreamForking\Command\ForkContentStream;
-use Neos\ContentRepository\Core\Feature\ContentStreamRemoval\Command\RemoveContentStream;
-
-/**
- * INTERNALS. Only to be used from WorkspaceCommandHandler!!!
- *
- * @internal from userland, you'll use ContentRepository::handle to dispatch commands
- * FIXME try to fully get rid of this handler :D and the external commands!
- */
-final class ContentStreamCommandHandler implements CommandHandlerInterface
-{
-    use ContentStreamHandling;
-
-    public function canHandle(CommandInterface $command): bool
-    {
-        return method_exists($this, 'handle' . (new \ReflectionClass($command))->getShortName());
-    }
-
-    public function handle(CommandInterface $command, CommandHandlingDependencies $commandHandlingDependencies): EventsToPublish
-    {
-        return match ($command::class) {
-            CloseContentStream::class => $this->handleCloseContentStream($command, $commandHandlingDependencies),
-            ReopenContentStream::class => $this->handleReopenContentStream($command, $commandHandlingDependencies),
-            RemoveContentStream::class => $this->handleRemoveContentStream($command, $commandHandlingDependencies),
-            default => throw new \DomainException('Cannot handle commands of class ' . get_class($command), 1710408206),
-        };
-    }
-
-    private function handleCloseContentStream(
-        CloseContentStream $command,
-        CommandHandlingDependencies $commandHandlingDependencies,
-    ): EventsToPublish {
-        return $this->closeContentStream($command->contentStreamId, $commandHandlingDependencies);
-    }
-
-    private function handleReopenContentStream(
-        ReopenContentStream $command,
-        CommandHandlingDependencies $commandHandlingDependencies,
-    ): EventsToPublish {
-        return $this->reopenContentStream($command->contentStreamId, $command->previousState, $commandHandlingDependencies);
-    }
-
-    private function handleRemoveContentStream(
-        RemoveContentStream $command,
-        CommandHandlingDependencies $commandHandlingDependencies
-    ): EventsToPublish {
-        return $this->removeContentStream($command->contentStreamId, $commandHandlingDependencies);
-    }
-}
diff --git a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamEventStreamName.php b/Neos.ContentRepository.Core/Classes/Feature/ContentStreamEventStreamName.php
index 5e3a7acad57..c0cdbb0a4f3 100644
--- a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamEventStreamName.php
+++ b/Neos.ContentRepository.Core/Classes/Feature/ContentStreamEventStreamName.php
@@ -24,7 +24,7 @@
  */
 final readonly class ContentStreamEventStreamName
 {
-    private const EVENT_STREAM_NAME_PREFIX = 'ContentStream:';
+    public const EVENT_STREAM_NAME_PREFIX = 'ContentStream:';
 
     private function __construct(
         public string $value
diff --git a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamHandling.php b/Neos.ContentRepository.Core/Classes/Feature/ContentStreamHandling.php
index 16d191cfc10..c3027d71147 100644
--- a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamHandling.php
+++ b/Neos.ContentRepository.Core/Classes/Feature/ContentStreamHandling.php
@@ -17,7 +17,6 @@
 use Neos\ContentRepository\Core\SharedModel\Exception\ContentStreamIsClosed;
 use Neos\ContentRepository\Core\SharedModel\Exception\ContentStreamIsNotClosed;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\EventStore\Model\EventStream\ExpectedVersion;
 
 trait ContentStreamHandling
@@ -74,12 +73,10 @@ private function closeContentStream(
 
     /**
      * @param ContentStreamId $contentStreamId The id of the content stream to reopen
-     * @param ContentStreamStatus $previousState The state the content stream was in before closing and is to be reset to
      * @phpstan-pure this method is pure, to persist the events they must be handled outside
      */
     private function reopenContentStream(
         ContentStreamId $contentStreamId,
-        ContentStreamStatus $previousState,
         CommandHandlingDependencies $commandHandlingDependencies,
     ): EventsToPublish {
         $this->requireContentStreamToExist($contentStreamId, $commandHandlingDependencies);
@@ -90,8 +87,7 @@ private function reopenContentStream(
             $streamName,
             Events::with(
                 new ContentStreamWasReopened(
-                    $contentStreamId,
-                    $previousState,
+                    $contentStreamId
                 ),
             ),
             ExpectedVersion::ANY()
@@ -197,7 +193,7 @@ private function requireContentStreamToNotBeClosed(
         ContentStreamId $contentStreamId,
         CommandHandlingDependencies $commandHandlingDependencies
     ): void {
-        if ($commandHandlingDependencies->getContentStreamStatus($contentStreamId) === ContentStreamStatus::CLOSED) {
+        if ($commandHandlingDependencies->isContentStreamClosed($contentStreamId)) {
             throw new ContentStreamIsClosed(
                 'Content stream "' . $contentStreamId->value . '" is closed.',
                 1710260081
@@ -209,7 +205,7 @@ private function requireContentStreamToBeClosed(
         ContentStreamId $contentStreamId,
         CommandHandlingDependencies $commandHandlingDependencies
     ): void {
-        if ($commandHandlingDependencies->getContentStreamStatus($contentStreamId) !== ContentStreamStatus::CLOSED) {
+        if (!$commandHandlingDependencies->isContentStreamClosed($contentStreamId)) {
             throw new ContentStreamIsNotClosed(
                 'Content stream "' . $contentStreamId->value . '" is not closed.',
                 1710405911
diff --git a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamRemoval/Command/RemoveContentStream.php b/Neos.ContentRepository.Core/Classes/Feature/ContentStreamRemoval/Command/RemoveContentStream.php
deleted file mode 100644
index 932a488a842..00000000000
--- a/Neos.ContentRepository.Core/Classes/Feature/ContentStreamRemoval/Command/RemoveContentStream.php
+++ /dev/null
@@ -1,44 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace Neos\ContentRepository\Core\Feature\ContentStreamRemoval\Command;
-
-/*
- * This file is part of the Neos.ContentRepository package.
- *
- * (c) Contributors of the Neos Project - www.neos.io
- *
- * This package is Open Source Software. For the full copyright and license
- * information, please view the LICENSE file which was distributed with this
- * source code.
- */
-
-use Neos\ContentRepository\Core\CommandHandler\CommandInterface;
-use Neos\ContentRepository\Core\Service\ContentStreamPruner;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-
-/**
- * Command to remove an existing content stream
- *
- * @internal implementation detail of the {@see ContentStreamPruner}. You must not use this command directly.
- * Direct use would throw away the content stream and may leave a broken state.
- */
-final readonly class RemoveContentStream implements CommandInterface
-{
-    /**
-     * @param ContentStreamId $contentStreamId The id of the content stream to remove
-     */
-    private function __construct(
-        public ContentStreamId $contentStreamId,
-    ) {
-    }
-
-    /**
-     * @param ContentStreamId $contentStreamId The id of the content stream to remove
-     */
-    public static function create(ContentStreamId $contentStreamId): self
-    {
-        return new self($contentStreamId);
-    }
-}
diff --git a/Neos.ContentRepository.Core/Classes/Feature/WorkspaceCommandHandler.php b/Neos.ContentRepository.Core/Classes/Feature/WorkspaceCommandHandler.php
index 0cea05dc7e3..9d4511ce755 100644
--- a/Neos.ContentRepository.Core/Classes/Feature/WorkspaceCommandHandler.php
+++ b/Neos.ContentRepository.Core/Classes/Feature/WorkspaceCommandHandler.php
@@ -58,7 +58,6 @@
 use Neos\ContentRepository\Core\SharedModel\Exception\WorkspaceDoesNotExist;
 use Neos\ContentRepository\Core\SharedModel\Exception\WorkspaceHasNoBaseWorkspaceName;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspace;
 use Neos\ContentRepository\Core\SharedModel\Workspace\WorkspaceName;
 use Neos\ContentRepository\Core\SharedModel\Workspace\WorkspaceStatus;
@@ -202,7 +201,6 @@ private function handlePublishWorkspace(
             // we have no changes, we just reopen; partial no-op
             yield $this->reopenContentStream(
                 $workspace->currentContentStreamId,
-                ContentStreamStatus::IN_USE_BY_WORKSPACE, // todo will be removed
                 $commandHandlingDependencies
             );
             return;
@@ -220,7 +218,6 @@ private function handlePublishWorkspace(
         } catch (WorkspaceRebaseFailed $workspaceRebaseFailed) {
             yield $this->reopenContentStream(
                 $workspace->currentContentStreamId,
-                ContentStreamStatus::IN_USE_BY_WORKSPACE, // todo will be removed
                 $commandHandlingDependencies
             );
             throw $workspaceRebaseFailed;
@@ -347,7 +344,6 @@ private function handleRebaseWorkspace(
         if (!$commandHandlingDependencies->contentStreamExists($workspace->currentContentStreamId)) {
             throw new \RuntimeException('Cannot rebase a workspace with a stateless content stream', 1711718314);
         }
-        $currentWorkspaceContentStreamState = $commandHandlingDependencies->getContentStreamStatus($workspace->currentContentStreamId);
 
         if (
             $workspace->status === WorkspaceStatus::UP_TO_DATE
@@ -396,7 +392,6 @@ static function ($handle) use ($rebaseableCommands): void {
         ) {
             yield $this->reopenContentStream(
                 $workspace->currentContentStreamId,
-                $currentWorkspaceContentStreamState,
                 $commandHandlingDependencies
             );
 
@@ -452,7 +447,6 @@ private function handlePublishIndividualNodesFromWorkspace(
         if (!$commandHandlingDependencies->contentStreamExists($workspace->currentContentStreamId)) {
             throw new \RuntimeException('Cannot publish nodes on a workspace with a stateless content stream', 1710410114);
         }
-        $currentWorkspaceContentStreamState = $commandHandlingDependencies->getContentStreamStatus($workspace->currentContentStreamId);
         $baseWorkspace = $this->requireBaseWorkspace($workspace, $commandHandlingDependencies);
         $this->requireContentStreamToNotBeClosed($baseWorkspace->currentContentStreamId, $commandHandlingDependencies);
         $baseContentStreamVersion = $commandHandlingDependencies->getContentStreamVersion($baseWorkspace->currentContentStreamId);
@@ -475,7 +469,6 @@ private function handlePublishIndividualNodesFromWorkspace(
             // almost a noop (e.g. random node ids were specified) ;)
             yield $this->reopenContentStream(
                 $workspace->currentContentStreamId,
-                $currentWorkspaceContentStreamState,
                 $commandHandlingDependencies
             );
             return;
@@ -496,7 +489,6 @@ private function handlePublishIndividualNodesFromWorkspace(
             } catch (WorkspaceRebaseFailed $workspaceRebaseFailed) {
                 yield $this->reopenContentStream(
                     $workspace->currentContentStreamId,
-                    ContentStreamStatus::IN_USE_BY_WORKSPACE, // todo will be removed
                     $commandHandlingDependencies
                 );
                 throw $workspaceRebaseFailed;
@@ -521,7 +513,6 @@ static function ($handle) use ($commandSimulator, $matchingCommands, $remainingC
         if ($commandSimulator->hasCommandsThatFailed()) {
             yield $this->reopenContentStream(
                 $workspace->currentContentStreamId,
-                $currentWorkspaceContentStreamState,
                 $commandHandlingDependencies
             );
 
@@ -590,7 +581,6 @@ private function handleDiscardIndividualNodesFromWorkspace(
         if (!$commandHandlingDependencies->contentStreamExists($workspace->currentContentStreamId)) {
             throw new \RuntimeException('Cannot discard nodes on a workspace with a stateless content stream', 1710408112);
         }
-        $currentWorkspaceContentStreamState = $commandHandlingDependencies->getContentStreamStatus($workspace->currentContentStreamId);
         $baseWorkspace = $this->requireBaseWorkspace($workspace, $commandHandlingDependencies);
 
         yield $this->closeContentStream(
@@ -611,7 +601,6 @@ private function handleDiscardIndividualNodesFromWorkspace(
             // if we have nothing to discard, we can just keep all. (e.g. random node ids were specified) It's almost a noop ;)
             yield $this->reopenContentStream(
                 $workspace->currentContentStreamId,
-                $currentWorkspaceContentStreamState,
                 $commandHandlingDependencies
             );
             return;
@@ -641,7 +630,6 @@ static function ($handle) use ($commandsToKeep): void {
         if ($commandSimulator->hasCommandsThatFailed()) {
             yield $this->reopenContentStream(
                 $workspace->currentContentStreamId,
-                $currentWorkspaceContentStreamState,
                 $commandHandlingDependencies
             );
             throw WorkspaceRebaseFailed::duringDiscard($commandSimulator->getCommandsThatFailed());
@@ -819,8 +807,7 @@ private function forkNewContentStreamAndApplyEvents(
             $eventsToApplyOnNewContentStream->withAppendedEvents(
                 Events::with(
                     new ContentStreamWasReopened(
-                        $newContentStreamId,
-                        ContentStreamStatus::IN_USE_BY_WORKSPACE // todo remove just temporary
+                        $newContentStreamId
                     )
                 )
             ),
diff --git a/Neos.ContentRepository.Core/Classes/Feature/WorkspaceEventStreamName.php b/Neos.ContentRepository.Core/Classes/Feature/WorkspaceEventStreamName.php
index 8d873a2a96a..1a14bf70895 100644
--- a/Neos.ContentRepository.Core/Classes/Feature/WorkspaceEventStreamName.php
+++ b/Neos.ContentRepository.Core/Classes/Feature/WorkspaceEventStreamName.php
@@ -14,7 +14,7 @@
  */
 final readonly class WorkspaceEventStreamName
 {
-    private const EVENT_STREAM_NAME_PREFIX = 'Workspace:';
+    public const EVENT_STREAM_NAME_PREFIX = 'Workspace:';
 
     private function __construct(
         public string $eventStreamName,
diff --git a/Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner.php b/Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner.php
index 94f745e0a10..5afbcddc590 100644
--- a/Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner.php
+++ b/Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner.php
@@ -5,17 +5,34 @@
 namespace Neos\ContentRepository\Core\Service;
 
 use Neos\ContentRepository\Core\ContentRepository;
+use Neos\ContentRepository\Core\EventStore\EventNormalizer;
 use Neos\ContentRepository\Core\Factory\ContentRepositoryServiceInterface;
+use Neos\ContentRepository\Core\Feature\ContentStreamCreation\Event\ContentStreamWasCreated;
 use Neos\ContentRepository\Core\Feature\ContentStreamEventStreamName;
-use Neos\ContentRepository\Core\Feature\ContentStreamRemoval\Command\RemoveContentStream;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStream;
+use Neos\ContentRepository\Core\Feature\ContentStreamForking\Event\ContentStreamWasForked;
+use Neos\ContentRepository\Core\Feature\ContentStreamRemoval\Event\ContentStreamWasRemoved;
+use Neos\ContentRepository\Core\Feature\WorkspaceCreation\Event\RootWorkspaceWasCreated;
+use Neos\ContentRepository\Core\Feature\WorkspaceCreation\Event\WorkspaceWasCreated;
+use Neos\ContentRepository\Core\Feature\WorkspaceEventStreamName;
+use Neos\ContentRepository\Core\Feature\WorkspacePublication\Event\WorkspaceWasDiscarded;
+use Neos\ContentRepository\Core\Feature\WorkspacePublication\Event\WorkspaceWasPartiallyDiscarded;
+use Neos\ContentRepository\Core\Feature\WorkspacePublication\Event\WorkspaceWasPartiallyPublished;
+use Neos\ContentRepository\Core\Feature\WorkspacePublication\Event\WorkspaceWasPublished;
+use Neos\ContentRepository\Core\Feature\WorkspaceRebase\Event\WorkspaceRebaseFailed;
+use Neos\ContentRepository\Core\Feature\WorkspaceRebase\Event\WorkspaceWasRebased;
+use Neos\ContentRepository\Core\Service\ContentStreamPruner\ContentStreamForPruning;
+use Neos\ContentRepository\Core\Service\ContentStreamPruner\ContentStreamStatus;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreams;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\EventStore\EventStoreInterface;
+use Neos\EventStore\Model\Event\EventType;
+use Neos\EventStore\Model\Event\EventTypes;
+use Neos\EventStore\Model\Event\StreamName;
+use Neos\EventStore\Model\EventStream\EventStreamFilter;
+use Neos\EventStore\Model\EventStream\ExpectedVersion;
+use Neos\EventStore\Model\EventStream\VirtualStreamName;
 
 /**
- * For implementation details of the content stream states and removed state, see {@see ContentStream}.
+ * For implementation details of the content stream states and removed state, see {@see ContentStreamForPruning}.
  *
  * @api
  */
@@ -24,86 +41,194 @@ class ContentStreamPruner implements ContentRepositoryServiceInterface
     public function __construct(
         private readonly ContentRepository $contentRepository,
         private readonly EventStoreInterface $eventStore,
+        private readonly EventNormalizer $eventNormalizer
     ) {
     }
 
     /**
-     * Remove all content streams which are not needed anymore from the projections.
+     * Detects if dangling content streams exists and which content streams could be pruned from the event stream
      *
-     * NOTE: This still **keeps** the event stream as is; so it would be possible to re-construct the content stream
-     *       at a later point in time (though we currently do not provide any API for it).
+     * Dangling content streams
+     * ------------------------
      *
-     *       To remove the deleted Content Streams,
-     *       call {@see ContentStreamPruner::pruneRemovedFromEventStream()} afterwards.
+     * Content streams that are not removed via the event ContentStreamWasRemoved and are not in use by a workspace
+     * (not a current's workspace content stream).
      *
-     * By default, only content streams in STATE_NO_LONGER_IN_USE and STATE_REBASE_ERROR will be removed.
-     * If you also call with $removeTemporary=true, will delete ALL content streams which are currently not assigned
-     * to a workspace (f.e. dangling ones in FORKED or CREATED.).
+     * Previously before Neos 9 beta 15 (#5301), dangling content streams were not removed during publishing, discard or rebase.
      *
-     * @param bool $removeTemporary if TRUE, will delete ALL content streams not bound to a workspace
-     * @return iterable<int,ContentStreamId> the identifiers of the removed content streams
+     * {@see removeDanglingContentStreams}
+     *
+     * Pruneable content streams
+     * -------------------------
+     *
+     * Content streams that were removed ContentStreamWasRemoved e.g. after publishing, and are not required for a full
+     * replay to reconstruct the current projections state. The ability to reconstitute a previous state will be lost.
+     *
+     * {@see pruneRemovedFromEventStream}
+     *
+     * @return bool false if dangling content streams exist because they should not
      */
-    public function prune(bool $removeTemporary = false): iterable
+    public function outputStatus(\Closure $outputFn): bool
     {
-        $status = [ContentStreamStatus::NO_LONGER_IN_USE, ContentStreamStatus::REBASE_ERROR];
-        if ($removeTemporary) {
-            $status[] = ContentStreamStatus::CREATED;
-            $status[] = ContentStreamStatus::FORKED;
-        }
-        $unusedContentStreams = $this->contentRepository->findContentStreams()->filter(
-            static fn (ContentStream $contentStream) => in_array($contentStream->status, $status, true),
-        );
-        $unusedContentStreamIds = [];
-        foreach ($unusedContentStreams as $contentStream) {
-            if ($contentStream->removed) {
+        $allContentStreams = $this->findAllContentStreams();
+
+        $danglingContentStreamPresent = false;
+        foreach ($allContentStreams as $contentStream) {
+            if (!$contentStream->isDangling()) {
                 continue;
             }
-            $this->contentRepository->handle(
-                RemoveContentStream::create($contentStream->id)
-            );
-            $unusedContentStreamIds[] = $contentStream->id;
+            if ($danglingContentStreamPresent === false) {
+                $outputFn(sprintf('Dangling content streams that are not removed (ContentStreamWasRemoved) and not %s:', ContentStreamStatus::IN_USE_BY_WORKSPACE->value));
+            }
+
+            if ($contentStream->status->isTemporary()) {
+                $outputFn(sprintf('  id: %s temporary %s at %s', $contentStream->id->value, $contentStream->status->value, $contentStream->created->format('Y-m-d H:i')));
+            } else {
+                $outputFn(sprintf('  id: %s %s', $contentStream->id->value, $contentStream->status->value));
+            }
+
+            $danglingContentStreamPresent = true;
+        }
+
+        if ($danglingContentStreamPresent === true) {
+            $outputFn('To remove the dangling streams from the projections please run ./flow contentStream:removeDangling');
+            $outputFn('Then they are ready for removal from the event stream');
+            $outputFn();
+        } else {
+            $outputFn('Okay. No dangling streams found');
+            $outputFn();
+        }
+
+        $pruneableContentStreams = $this->findRemovedContentStreamsThatAreUnused($allContentStreams);
+
+        $pruneableContentStreamPresent = false;
+        foreach ($pruneableContentStreams as $pruneableContentStream) {
+            if ($pruneableContentStreamPresent === false) {
+                $outputFn('Removed content streams that can be pruned from the event stream');
+            }
+            $pruneableContentStreamPresent = true;
+            $outputFn(sprintf('  id: %s previous state: %s', $pruneableContentStream->id->value, $pruneableContentStream->status->value));
+        }
+
+        if ($pruneableContentStreamPresent === true) {
+            $outputFn('To prune the removed streams from the event stream run ./flow contentStream:pruneRemovedFromEventstream');
+        } else {
+            $outputFn('Okay. No pruneable streams in the event stream');
         }
 
-        return $unusedContentStreamIds;
+        return !$danglingContentStreamPresent;
     }
 
     /**
-     * Remove unused and deleted content streams from the event stream; effectively REMOVING information completely.
+     * Removes all nodes, hierarchy relations and content stream entries which are not needed anymore from the projections.
      *
-     * This is not so easy for nested workspaces / content streams:
-     *   - As long as content streams are used as basis for others which are IN_USE_BY_WORKSPACE,
-     *     these dependent Content Streams are not allowed to be removed in the event store.
+     * NOTE: This still **keeps** the event stream as is; so it would be possible to re-construct the content stream at a later point in time.
      *
-     *   - Otherwise, we cannot replay the other content streams correctly (if the base content streams are missing).
+     * To prune the removed content streams from the event stream, call {@see ContentStreamPruner::pruneRemovedFromEventStream()} afterwards.
      *
-     * @return ContentStreams the removed content streams
+     * @param \DateTimeImmutable $removeTemporaryBefore includes all temporary content streams like FORKED or CREATED older than that in the removal
      */
-    public function pruneRemovedFromEventStream(): ContentStreams
+    public function removeDanglingContentStreams(\Closure $outputFn, \DateTimeImmutable $removeTemporaryBefore): void
     {
-        $removedContentStreams = $this->findUnusedAndRemovedContentStreams();
-        foreach ($removedContentStreams as $removedContentStream) {
-            $streamName = ContentStreamEventStreamName::fromContentStreamId($removedContentStream->id)
-                ->getEventStreamName();
-            $this->eventStore->deleteStream($streamName);
+        $allContentStreams = $this->findAllContentStreams();
+
+        $danglingContentStreamsPresent = false;
+        foreach ($allContentStreams as $contentStream) {
+            if (!$contentStream->isDangling()) {
+                continue;
+            }
+            if (
+                $contentStream->status->isTemporary()
+                && $removeTemporaryBefore < $contentStream->created
+            ) {
+                $outputFn(sprintf('Did not remove %s temporary %s at %s', $contentStream->id->value, $contentStream->status->value, $contentStream->created->format('Y-m-d H:i')));
+                continue;
+            }
+
+            $this->eventStore->commit(
+                ContentStreamEventStreamName::fromContentStreamId($contentStream->id)->getEventStreamName(),
+                $this->eventNormalizer->normalize(
+                    new ContentStreamWasRemoved(
+                        $contentStream->id
+                    )
+                ),
+                ExpectedVersion::STREAM_EXISTS()
+            );
+
+            $outputFn(sprintf('Removed %s with status %s', $contentStream->id, $contentStream->status->value));
+
+            $danglingContentStreamsPresent = true;
+        }
+
+        if ($danglingContentStreamsPresent) {
+            try {
+                $this->contentRepository->catchUpProjections();
+            } catch (\Throwable $e) {
+                $outputFn(sprintf('Could not catchup after removing unused content streams: %s. You might need to use ./flow contentstream:pruneremovedfromeventstream and replay.', $e->getMessage()));
+            }
+        } else {
+            $outputFn('Okay. No pruneable streams in the event stream');
         }
-        return $removedContentStreams;
     }
 
-    public function pruneAll(): void
+    /**
+     * Prune removed content streams that are unused from the event stream; effectively REMOVING information completely.
+     *
+     * Note that replaying to only a previous point in time would not be possible anymore as workspace would reference non-existing content streams.
+     *
+     * @see findRemovedContentStreamsThatAreUnused for implementation
+     */
+    public function pruneRemovedFromEventStream(\Closure $outputFn): void
     {
-        foreach ($this->contentRepository->findContentStreams() as $contentStream) {
-            $streamName = ContentStreamEventStreamName::fromContentStreamId($contentStream->id)->getEventStreamName();
-            $this->eventStore->deleteStream($streamName);
+        $allContentStreams = $this->findAllContentStreams();
+
+        $pruneableContentStreams = $this->findRemovedContentStreamsThatAreUnused($allContentStreams);
+
+        $pruneableContentStreamsPresent = false;
+        foreach ($pruneableContentStreams as $pruneableContentStream) {
+            $this->eventStore->deleteStream(
+                ContentStreamEventStreamName::fromContentStreamId(
+                    $pruneableContentStream->id
+                )->getEventStreamName()
+            );
+            $pruneableContentStreamsPresent = true;
+            $outputFn(sprintf('Removed events for %s', $pruneableContentStream->id->value));
+        }
+
+        if ($pruneableContentStreamsPresent === false) {
+            $outputFn('Okay. There are no pruneable content streams.');
         }
     }
 
-    private function findUnusedAndRemovedContentStreams(): ContentStreams
+    public function pruneAllWorkspacesAndContentStreamsFromEventStream(): void
     {
-        $allContentStreams = $this->contentRepository->findContentStreams();
+        foreach ($this->findAllContentStreamStreamNames() as $contentStreamStreamName) {
+            $this->eventStore->deleteStream($contentStreamStreamName);
+        }
+        foreach ($this->findAllWorkspaceStreamNames() as $workspaceStreamName) {
+            $this->eventStore->deleteStream($workspaceStreamName);
+        }
+    }
 
+    /**
+     * Find all removed content streams that are unused in the event stream
+     *
+     * This is not so easy for nested workspaces / content streams:
+     * - As long as content streams are used as basis for others which are IN_USE_BY_WORKSPACE,
+     *   these dependent Content Streams are not allowed to be removed in the event stream.
+     * - Otherwise, we cannot replay the other content streams correctly (if the base content streams are missing).
+     *
+     * @param array<string, ContentStreamForPruning> $allContentStreams
+     * @return list<ContentStreamForPruning>
+     */
+    private function findRemovedContentStreamsThatAreUnused(array $allContentStreams): array
+    {
         /** @var array<string,bool> $transitiveUsedStreams */
         $transitiveUsedStreams = [];
-        /** @var list<ContentStreamId> $contentStreamIdsStack */
+        /**
+         * Collection of content streams we iterate through to build up all streams that are in use transitively (by being a source content stream) or because it is in use
+         * @var list<ContentStreamId> $contentStreamIdsStack
+         */
         $contentStreamIdsStack = [];
 
         // Step 1: Find all content streams currently in direct use by a workspace
@@ -132,13 +257,202 @@ private function findUnusedAndRemovedContentStreams(): ContentStreams
         }
 
         // Step 3: Check for removed content streams which we do not need anymore transitively
-        $removedContentStreams = [];
+        $removedContentStreamsThatAreUnused = [];
         foreach ($allContentStreams as $contentStream) {
             if ($contentStream->removed && !array_key_exists($contentStream->id->value, $transitiveUsedStreams)) {
-                $removedContentStreams[] = $contentStream;
+                $removedContentStreamsThatAreUnused[] = $contentStream;
+            }
+        }
+
+        return $removedContentStreamsThatAreUnused;
+    }
+
+    /**
+     * @return array<string, ContentStreamForPruning>
+     */
+    private function findAllContentStreams(): array
+    {
+        $events = $this->eventStore->load(
+            VirtualStreamName::forCategory(ContentStreamEventStreamName::EVENT_STREAM_NAME_PREFIX),
+            EventStreamFilter::create(
+                EventTypes::create(
+                    EventType::fromString('ContentStreamWasCreated'),
+                    EventType::fromString('ContentStreamWasForked'),
+                    EventType::fromString('ContentStreamWasRemoved'),
+                )
+            )
+        );
+
+        /** @var array<string,ContentStreamForPruning> $cs */
+        $cs = [];
+        foreach ($events as $eventEnvelope) {
+            $domainEvent = $this->eventNormalizer->denormalize($eventEnvelope->event);
+
+            switch ($domainEvent::class) {
+                case ContentStreamWasCreated::class:
+                    $cs[$domainEvent->contentStreamId->value] = ContentStreamForPruning::create(
+                        $domainEvent->contentStreamId,
+                        ContentStreamStatus::CREATED,
+                        null,
+                        $eventEnvelope->recordedAt
+                    );
+                    break;
+                case ContentStreamWasForked::class:
+                    $cs[$domainEvent->newContentStreamId->value] = ContentStreamForPruning::create(
+                        $domainEvent->newContentStreamId,
+                        ContentStreamStatus::FORKED,
+                        $domainEvent->sourceContentStreamId,
+                        $eventEnvelope->recordedAt
+                    );
+                    break;
+                case ContentStreamWasRemoved::class:
+                    if (isset($cs[$domainEvent->contentStreamId->value])) {
+                        $cs[$domainEvent->contentStreamId->value] = $cs[$domainEvent->contentStreamId->value]
+                            ->withRemoved();
+                    }
+                    break;
+                default:
+                    throw new \RuntimeException(sprintf('Unhandled event %s', $eventEnvelope->event->type->value));
+            }
+        }
+
+        $workspaceEvents = $this->eventStore->load(
+            VirtualStreamName::forCategory(WorkspaceEventStreamName::EVENT_STREAM_NAME_PREFIX),
+            EventStreamFilter::create(
+                EventTypes::create(
+                    EventType::fromString('RootWorkspaceWasCreated'),
+                    EventType::fromString('WorkspaceWasCreated'),
+                    EventType::fromString('WorkspaceWasDiscarded'),
+                    EventType::fromString('WorkspaceWasPartiallyDiscarded'),
+                    EventType::fromString('WorkspaceWasPartiallyPublished'),
+                    EventType::fromString('WorkspaceWasPublished'),
+                    EventType::fromString('WorkspaceWasRebased'),
+                    EventType::fromString('WorkspaceRebaseFailed'),
+                    // we don't need to track WorkspaceWasRemoved as a ContentStreamWasRemoved event would be emitted before
+                )
+            )
+        );
+        foreach ($workspaceEvents as $eventEnvelope) {
+            $domainEvent = $this->eventNormalizer->denormalize($eventEnvelope->event);
+
+            switch ($domainEvent::class) {
+                case RootWorkspaceWasCreated::class:
+                    if (isset($cs[$domainEvent->newContentStreamId->value])) {
+                        $cs[$domainEvent->newContentStreamId->value] = $cs[$domainEvent->newContentStreamId->value]
+                                ->withStatus(ContentStreamStatus::IN_USE_BY_WORKSPACE);
+                    }
+                    break;
+                case WorkspaceWasCreated::class:
+                    if (isset($cs[$domainEvent->newContentStreamId->value])) {
+                        $cs[$domainEvent->newContentStreamId->value] = $cs[$domainEvent->newContentStreamId->value]
+                                ->withStatus(ContentStreamStatus::IN_USE_BY_WORKSPACE);
+                    }
+                    break;
+                case WorkspaceWasDiscarded::class:
+                    if (isset($cs[$domainEvent->newContentStreamId->value])) {
+                        $cs[$domainEvent->newContentStreamId->value] = $cs[$domainEvent->newContentStreamId->value]
+                            ->withStatus(ContentStreamStatus::IN_USE_BY_WORKSPACE);
+                    }
+                    if (isset($cs[$domainEvent->previousContentStreamId->value])) {
+                        $cs[$domainEvent->previousContentStreamId->value] = $cs[$domainEvent->previousContentStreamId->value]
+                            ->withStatus(ContentStreamStatus::NO_LONGER_IN_USE);
+                    }
+                    break;
+                case WorkspaceWasPartiallyDiscarded::class:
+                    if (isset($cs[$domainEvent->newContentStreamId->value])) {
+                        $cs[$domainEvent->newContentStreamId->value] = $cs[$domainEvent->newContentStreamId->value]
+                            ->withStatus(ContentStreamStatus::IN_USE_BY_WORKSPACE);
+                    }
+                    if (isset($cs[$domainEvent->previousContentStreamId->value])) {
+                        $cs[$domainEvent->previousContentStreamId->value] = $cs[$domainEvent->previousContentStreamId->value]
+                            ->withStatus(ContentStreamStatus::NO_LONGER_IN_USE);
+                    }
+                    break;
+                case WorkspaceWasPartiallyPublished::class:
+                    if (isset($cs[$domainEvent->newSourceContentStreamId->value])) {
+                        $cs[$domainEvent->newSourceContentStreamId->value] = $cs[$domainEvent->newSourceContentStreamId->value]
+                            ->withStatus(ContentStreamStatus::IN_USE_BY_WORKSPACE);
+                    }
+                    if (isset($cs[$domainEvent->previousSourceContentStreamId->value])) {
+                        $cs[$domainEvent->previousSourceContentStreamId->value] = $cs[$domainEvent->previousSourceContentStreamId->value]
+                            ->withStatus(ContentStreamStatus::NO_LONGER_IN_USE);
+                    }
+                    break;
+                case WorkspaceWasPublished::class:
+                    if (isset($cs[$domainEvent->newSourceContentStreamId->value])) {
+                        $cs[$domainEvent->newSourceContentStreamId->value] = $cs[$domainEvent->newSourceContentStreamId->value]
+                            ->withStatus(ContentStreamStatus::IN_USE_BY_WORKSPACE);
+                    }
+                    if (isset($cs[$domainEvent->previousSourceContentStreamId->value])) {
+                        $cs[$domainEvent->previousSourceContentStreamId->value] = $cs[$domainEvent->previousSourceContentStreamId->value]
+                            ->withStatus(ContentStreamStatus::NO_LONGER_IN_USE);
+                    }
+                    break;
+                case WorkspaceWasRebased::class:
+                    if (isset($cs[$domainEvent->newContentStreamId->value])) {
+                        $cs[$domainEvent->newContentStreamId->value] = $cs[$domainEvent->newContentStreamId->value]
+                            ->withStatus(ContentStreamStatus::IN_USE_BY_WORKSPACE);
+                    }
+                    if (isset($cs[$domainEvent->previousContentStreamId->value])) {
+                        $cs[$domainEvent->previousContentStreamId->value] = $cs[$domainEvent->previousContentStreamId->value]
+                            ->withStatus(ContentStreamStatus::NO_LONGER_IN_USE);
+                    }
+                    break;
+                case WorkspaceRebaseFailed::class:
+                    // legacy handling, as we previously kept failed candidateContentStreamId we make it behave like a ContentStreamWasRemoved event to clean up:
+                    if (isset($cs[$domainEvent->candidateContentStreamId->value])) {
+                        $cs[$domainEvent->candidateContentStreamId->value] = $cs[$domainEvent->candidateContentStreamId->value]
+                            ->withRemoved();
+                    }
+                    break;
+                default:
+                    throw new \RuntimeException(sprintf('Unhandled event %s', $eventEnvelope->event->type->value));
             }
         }
+        return $cs;
+    }
+
+    /**
+     * @return list<StreamName>
+     */
+    private function findAllContentStreamStreamNames(): array
+    {
+        $events = $this->eventStore->load(
+            VirtualStreamName::forCategory(ContentStreamEventStreamName::EVENT_STREAM_NAME_PREFIX),
+            EventStreamFilter::create(
+                EventTypes::create(
+                    // we are only interested in the creation events to limit the amount of events to fetch
+                    EventType::fromString('ContentStreamWasCreated'),
+                    EventType::fromString('ContentStreamWasForked')
+                )
+            )
+        );
+        $allStreamNames = [];
+        foreach ($events as $eventEnvelope) {
+            $allStreamNames[] = $eventEnvelope->streamName;
+        }
+        return array_unique($allStreamNames, SORT_REGULAR);
+    }
 
-        return ContentStreams::fromArray($removedContentStreams);
+    /**
+     * @return list<StreamName>
+     */
+    private function findAllWorkspaceStreamNames(): array
+    {
+        $events = $this->eventStore->load(
+            VirtualStreamName::forCategory(WorkspaceEventStreamName::EVENT_STREAM_NAME_PREFIX),
+            EventStreamFilter::create(
+                EventTypes::create(
+                    // we are only interested in the creation events to limit the amount of events to fetch
+                    EventType::fromString('RootWorkspaceWasCreated'),
+                    EventType::fromString('WorkspaceWasCreated')
+                )
+            )
+        );
+        $allStreamNames = [];
+        foreach ($events as $eventEnvelope) {
+            $allStreamNames[] = $eventEnvelope->streamName;
+        }
+        return array_unique($allStreamNames, SORT_REGULAR);
     }
 }
diff --git a/Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner/ContentStreamForPruning.php b/Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner/ContentStreamForPruning.php
new file mode 100644
index 00000000000..9264fc8b9f7
--- /dev/null
+++ b/Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner/ContentStreamForPruning.php
@@ -0,0 +1,111 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Neos\ContentRepository\Core\Service\ContentStreamPruner;
+
+use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
+
+/**
+ * This model reflects if content streams are currently in use or not. Each content stream
+ * is first CREATED or FORKED, and then moves to the IN_USE or REBASE_ERROR states; or is removed directly
+ * in case of temporary content streams.
+ *
+ * FORKING: Content streams are forked from a base content stream. It can happen that the base content
+ * stream is NO_LONGER_IN_USE, but the child content stream is still IN_USE_BY_WORKSPACE. In this case,
+ * the base content stream can go to removed=true (removed from the content graph), but needs to be retained
+ * in the event store: If we do a full replay, we need the events of the base content stream before the
+ * fork happened to rebuild the child content stream.
+ * This logic is done in {@see ContentStreamPruner::findUnusedAndRemovedContentStreamIds()}.
+ *
+ * TEMPORARY content streams: Projections should take care to dispose their temporary content streams,
+ * by triggering a ContentStreamWasRemoved event after the content stream is no longer used.
+ *
+ * The different status a content stream can be in
+ *
+ *            │                       │
+ *            │(for root              │during
+ *            │ content               │rebase
+ *            ▼ stream)               ▼
+ *      ┌──────────┐            ┌──────────┐             Temporary
+ *      │ CREATED  │            │  FORKED  │────┐          status
+ *      └──────────┘            └──────────┘    for
+ *            │                       │      temporary
+ *            ├───────────────────────┤       content
+ *            ▼                       │       streams
+ *  ┌───────────────────┐             │         │
+ *  │IN_USE_BY_WORKSPACE│             │         │
+ *  └───────────────────┘             │         │        Persistent
+ *            │                       │         │          status
+ *            ▼                       │         │
+ *  ┌───────────────────┐             │         │
+ *  │ NO_LONGER_IN_USE  │             │         │
+ *  └───────────────────┘             │         │
+ *            │                       │         │
+ *            └──────────┬────────────┘         │
+ *                       ▼                      │
+ *  ┌────────────────────────────────────────┐  │
+ *  │              removed=true              │  │
+ *  │     => removed from content graph      │◀─┘
+ *  └────────────────────────────────────────┘           Cleanup
+ *                       │
+ *                       ▼
+ *  ┌────────────────────────────────────────┐
+ *  │  completely deleted from event stream  │
+ *  └────────────────────────────────────────┘
+ *
+ * @internal
+ */
+final readonly class ContentStreamForPruning
+{
+    private function __construct(
+        public ContentStreamId $id,
+        public ContentStreamStatus $status,
+        public ?ContentStreamId $sourceContentStreamId,
+        public \DateTimeImmutable $created,
+        public bool $removed,
+    ) {
+    }
+
+    public static function create(
+        ContentStreamId $id,
+        ContentStreamStatus $status,
+        ?ContentStreamId $sourceContentStreamId,
+        \DateTimeImmutable $create,
+    ): self {
+        return new self(
+            $id,
+            $status,
+            $sourceContentStreamId,
+            $create,
+            false
+        );
+    }
+
+    public function withStatus(ContentStreamStatus $status): self
+    {
+        return new self(
+            $this->id,
+            $status,
+            $this->sourceContentStreamId,
+            $this->created,
+            $this->removed
+        );
+    }
+
+    public function withRemoved(): self
+    {
+        return new self(
+            $this->id,
+            $this->status,
+            $this->sourceContentStreamId,
+            $this->created,
+            true
+        );
+    }
+
+    public function isDangling(): bool
+    {
+        return !$this->removed && $this->status !== ContentStreamStatus::IN_USE_BY_WORKSPACE;
+    }
+}
diff --git a/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/ContentStream.monopic b/Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner/ContentStreamStatus.monopic
similarity index 100%
rename from Neos.ContentRepository.Core/Classes/SharedModel/Workspace/ContentStream.monopic
rename to Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner/ContentStreamStatus.monopic
diff --git a/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/ContentStreamStatus.php b/Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner/ContentStreamStatus.php
similarity index 52%
rename from Neos.ContentRepository.Core/Classes/SharedModel/Workspace/ContentStreamStatus.php
rename to Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner/ContentStreamStatus.php
index ac21aeaa4cc..1c96d2f62e0 100644
--- a/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/ContentStreamStatus.php
+++ b/Neos.ContentRepository.Core/Classes/Service/ContentStreamPruner/ContentStreamStatus.php
@@ -12,19 +12,19 @@
 
 declare(strict_types=1);
 
-namespace Neos\ContentRepository\Core\SharedModel\Workspace;
+namespace Neos\ContentRepository\Core\Service\ContentStreamPruner;
 
 /**
  * @api
  */
-enum ContentStreamStatus: string implements \JsonSerializable
+enum ContentStreamStatus: string
 {
     /**
      * the content stream was created, but not yet assigned to a workspace.
      *
      * **temporary state** which should not appear if the system is idle (for content streams which are used with workspaces).
      */
-    case CREATED = 'CREATED';
+    case CREATED = 'created';
 
     /**
      * FORKED means the content stream was forked from an existing content stream, but not yet assigned
@@ -32,38 +32,23 @@ enum ContentStreamStatus: string implements \JsonSerializable
      *
      * **temporary state** which should not appear if the system is idle (for content streams which are used with workspaces).
      */
-    case FORKED = 'FORKED';
+    case FORKED = 'forked';
 
     /**
      * the content stream is currently referenced as the "active" content stream by a workspace.
      */
-    case IN_USE_BY_WORKSPACE = 'IN_USE_BY_WORKSPACE';
-
-    /**
-     * a workspace was tried to be rebased, and during the rebase an error occured. This is the content stream
-     * which contains the errored state - so that we can recover content from it (probably manually)
-     *
-     * @deprecated legacy status, FIXME clean up! https://github.com/neos/neos-development-collection/issues/5101
-     */
-    case REBASE_ERROR = 'REBASE_ERROR';
-
-    /**
-     * the content stream was closed and must no longer accept new events
-     */
-    case CLOSED = 'CLOSED';
+    case IN_USE_BY_WORKSPACE = 'in use by workspace';
 
     /**
      * the content stream is not used anymore, and can be removed.
      */
-    case NO_LONGER_IN_USE = 'NO_LONGER_IN_USE';
-
-    public static function fromString(string $value): self
-    {
-        return self::from($value);
-    }
+    case NO_LONGER_IN_USE = 'no longer in use';
 
-    public function jsonSerialize(): string
+    public function isTemporary(): bool
     {
-        return $this->value;
+        return match ($this) {
+            self::CREATED, self::FORKED => true,
+            default => false
+        };
     }
 }
diff --git a/Neos.ContentRepository.Core/Classes/Service/ContentStreamPrunerFactory.php b/Neos.ContentRepository.Core/Classes/Service/ContentStreamPrunerFactory.php
index 66acf840a9e..f9940f8f56a 100644
--- a/Neos.ContentRepository.Core/Classes/Service/ContentStreamPrunerFactory.php
+++ b/Neos.ContentRepository.Core/Classes/Service/ContentStreamPrunerFactory.php
@@ -18,6 +18,7 @@ public function build(ContentRepositoryServiceFactoryDependencies $serviceFactor
         return new ContentStreamPruner(
             $serviceFactoryDependencies->contentRepository,
             $serviceFactoryDependencies->eventStore,
+            $serviceFactoryDependencies->eventNormalizer
         );
     }
 }
diff --git a/Neos.ContentRepository.Core/Classes/Service/WorkspaceMaintenanceService.php b/Neos.ContentRepository.Core/Classes/Service/WorkspaceMaintenanceService.php
index 0c2ee14feab..b166f0d1271 100644
--- a/Neos.ContentRepository.Core/Classes/Service/WorkspaceMaintenanceService.php
+++ b/Neos.ContentRepository.Core/Classes/Service/WorkspaceMaintenanceService.php
@@ -6,13 +6,11 @@
 
 use Neos\ContentRepository\Core\ContentRepository;
 use Neos\ContentRepository\Core\Factory\ContentRepositoryServiceInterface;
-use Neos\ContentRepository\Core\Feature\WorkspaceEventStreamName;
 use Neos\ContentRepository\Core\Feature\WorkspaceRebase\Command\RebaseWorkspace;
 use Neos\ContentRepository\Core\Feature\WorkspaceRebase\Dto\RebaseErrorHandlingStrategy;
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspace;
 use Neos\ContentRepository\Core\SharedModel\Workspace\Workspaces;
 use Neos\ContentRepository\Core\SharedModel\Workspace\WorkspaceStatus;
-use Neos\EventStore\EventStoreInterface;
 
 /**
  * @api
@@ -20,8 +18,7 @@
 class WorkspaceMaintenanceService implements ContentRepositoryServiceInterface
 {
     public function __construct(
-        private readonly ContentRepository $contentRepository,
-        private readonly EventStoreInterface $eventStore,
+        private readonly ContentRepository $contentRepository
     ) {
     }
 
@@ -33,7 +30,7 @@ public function rebaseOutdatedWorkspaces(?RebaseErrorHandlingStrategy $strategy
         $outdatedWorkspaces = $this->contentRepository->findWorkspaces()->filter(
             fn (Workspace $workspace) => $workspace->status === WorkspaceStatus::OUTDATED
         );
-        /** @var Workspace $workspace */
+        // todo we need to loop through the workspaces from root level first
         foreach ($outdatedWorkspaces as $workspace) {
             if ($workspace->status !== WorkspaceStatus::OUTDATED) {
                 continue;
@@ -49,12 +46,4 @@ public function rebaseOutdatedWorkspaces(?RebaseErrorHandlingStrategy $strategy
 
         return $outdatedWorkspaces;
     }
-
-    public function pruneAll(): void
-    {
-        foreach ($this->contentRepository->findWorkspaces() as $workspace) {
-            $streamName = WorkspaceEventStreamName::fromWorkspaceName($workspace->workspaceName)->getEventStreamName();
-            $this->eventStore->deleteStream($streamName);
-        }
-    }
 }
diff --git a/Neos.ContentRepository.Core/Classes/Service/WorkspaceMaintenanceServiceFactory.php b/Neos.ContentRepository.Core/Classes/Service/WorkspaceMaintenanceServiceFactory.php
index eef7d991618..1e2b9c2f9ff 100644
--- a/Neos.ContentRepository.Core/Classes/Service/WorkspaceMaintenanceServiceFactory.php
+++ b/Neos.ContentRepository.Core/Classes/Service/WorkspaceMaintenanceServiceFactory.php
@@ -17,8 +17,7 @@ public function build(
         ContentRepositoryServiceFactoryDependencies $serviceFactoryDependencies
     ): WorkspaceMaintenanceService {
         return new WorkspaceMaintenanceService(
-            $serviceFactoryDependencies->contentRepository,
-            $serviceFactoryDependencies->eventStore,
+            $serviceFactoryDependencies->contentRepository
         );
     }
 }
diff --git a/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/ContentStream.php b/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/ContentStream.php
index def543c48b0..c52109de269 100644
--- a/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/ContentStream.php
+++ b/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/ContentStream.php
@@ -14,72 +14,37 @@
 
 namespace Neos\ContentRepository\Core\SharedModel\Workspace;
 
-use Neos\ContentRepository\Core\Service\ContentStreamPruner;
 use Neos\EventStore\Model\Event\Version;
 
 /**
  * Content Stream Read Model
  *
- * This model reflects if content streams are currently in use or not. Each content stream
- * is first CREATED or FORKED, and then moves to the IN_USE or REBASE_ERROR states; or is removed directly
- * in case of temporary content streams.
- *
- * FORKING: Content streams are forked from a base content stream. It can happen that the base content
- * stream is NO_LONGER_IN_USE, but the child content stream is still IN_USE_BY_WORKSPACE. In this case,
- * the base content stream can go to removed=true (removed from the content graph), but needs to be retained
- * in the event store: If we do a full replay, we need the events of the base content stream before the
- * fork happened to rebuild the child content stream.
- * This logic is done in {@see ContentStreamPruner::findUnusedAndRemovedContentStreamIds()}.
- *
- * TEMPORARY content streams: Projections should take care to dispose their temporary content streams,
- * by triggering a ContentStreamWasRemoved event after the content stream is no longer used.
- *
- * The different status a content stream can be in
- *
- *            │                       │
- *            │(for root              │during
- *            │ content               │rebase
- *            ▼ stream)               ▼
- *      ┌──────────┐            ┌──────────┐             Temporary
- *      │ CREATED  │            │  FORKED  │────┐          status
- *      └──────────┘            └──────────┘    for
- *            │                       │      temporary
- *            ├───────────────────────┤       content
- *            ▼                       │       streams
- *  ┌───────────────────┐             │         │
- *  │IN_USE_BY_WORKSPACE│             │         │
- *  └───────────────────┘             │         │        Persistent
- *            │                       │         │          status
- *            ▼                       │         │
- *  ┌───────────────────┐             │         │
- *  │ NO_LONGER_IN_USE  │             │         │
- *  └───────────────────┘             │         │
- *            │                       │         │
- *            └──────────┬────────────┘         │
- *                       ▼                      │
- *  ┌────────────────────────────────────────┐  │
- *  │              removed=true              │  │
- *  │     => removed from content graph      │◀─┘
- *  └────────────────────────────────────────┘           Cleanup
- *                       │
- *                       ▼
- *  ┌────────────────────────────────────────┐
- *  │  completely deleted from event stream  │
- *  └────────────────────────────────────────┘
- *
- * @api
+ * @api Note: The constructor is not part of the public API
  */
 final readonly class ContentStream
 {
-    /**
-     * @internal
-     */
-    public function __construct(
+    private function __construct(
         public ContentStreamId $id,
         public ?ContentStreamId $sourceContentStreamId,
-        public ContentStreamStatus $status,
         public Version $version,
-        public bool $removed
+        public bool $isClosed
     ) {
     }
+
+    /**
+     * @internal
+     */
+    public static function create(
+        ContentStreamId $id,
+        ?ContentStreamId $sourceContentStreamId,
+        Version $version,
+        bool $isClosed
+    ): self {
+        return new self(
+            $id,
+            $sourceContentStreamId,
+            $version,
+            $isClosed
+        );
+    }
 }
diff --git a/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/Workspace.php b/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/Workspace.php
index 7f51c161123..b3fd69556fe 100644
--- a/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/Workspace.php
+++ b/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/Workspace.php
@@ -17,43 +17,34 @@
 /**
  * Workspace Read Model
  *
- * @api
+ * @api Note: The constructor is not part of the public API
  */
 final readonly class Workspace
 {
     /**
-     * @var WorkspaceName Workspace identifier, unique within one Content Repository instance
+     * @param WorkspaceName $workspaceName Workspace identifier, unique within one Content Repository instance
+     * @param WorkspaceName|null $baseWorkspaceName Workspace identifier of the base workspace (i.e. the target when publishing changes) – if null this instance is considered a root (aka public) workspace
+     * @param ContentStreamId $currentContentStreamId The Content Stream this workspace currently points to – usually it is set to a new, empty content stream after publishing/rebasing the workspace
+     * @param WorkspaceStatus $status The current status of this workspace
      */
-    public WorkspaceName $workspaceName;
-
-    /**
-     * @var WorkspaceName|null Workspace identifier of the base workspace (i.e. the target when publishing changes) – if null this instance is considered a root (aka public) workspace
-     */
-    public ?WorkspaceName $baseWorkspaceName;
-
-    /**
-     * The Content Stream this workspace currently points to – usually it is set to a new, empty content stream after publishing/rebasing the workspace
-     */
-    public ContentStreamId $currentContentStreamId;
-
-    /**
-     * The current status of this workspace
-     */
-    public WorkspaceStatus $status;
+    private function __construct(
+        public WorkspaceName $workspaceName,
+        public ?WorkspaceName $baseWorkspaceName,
+        public ContentStreamId $currentContentStreamId,
+        public WorkspaceStatus $status,
+    ) {
+    }
 
     /**
      * @internal
      */
-    public function __construct(
+    public static function create(
         WorkspaceName $workspaceName,
         ?WorkspaceName $baseWorkspaceName,
         ContentStreamId $currentContentStreamId,
         WorkspaceStatus $status,
-    ) {
-        $this->workspaceName = $workspaceName;
-        $this->baseWorkspaceName = $baseWorkspaceName;
-        $this->currentContentStreamId = $currentContentStreamId;
-        $this->status = $status;
+    ): self {
+        return new self($workspaceName, $baseWorkspaceName, $currentContentStreamId, $status);
     }
 
     /**
diff --git a/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/Workspaces.php b/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/Workspaces.php
index f6af24a86ec..be0969b591d 100644
--- a/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/Workspaces.php
+++ b/Neos.ContentRepository.Core/Classes/SharedModel/Workspace/Workspaces.php
@@ -21,7 +21,6 @@
  *
  * @api
  */
-
 final class Workspaces implements \IteratorAggregate, \Countable
 {
     /**
diff --git a/Neos.ContentRepository.TestSuite/Classes/Behavior/Features/Bootstrap/CRTestSuiteTrait.php b/Neos.ContentRepository.TestSuite/Classes/Behavior/Features/Bootstrap/CRTestSuiteTrait.php
index 061d97dbf40..18fddc56f68 100644
--- a/Neos.ContentRepository.TestSuite/Classes/Behavior/Features/Bootstrap/CRTestSuiteTrait.php
+++ b/Neos.ContentRepository.TestSuite/Classes/Behavior/Features/Bootstrap/CRTestSuiteTrait.php
@@ -15,24 +15,22 @@
 namespace Neos\ContentRepository\TestSuite\Behavior\Features\Bootstrap;
 
 use Behat\Behat\Hook\Scope\BeforeScenarioScope;
+use Behat\Gherkin\Node\PyStringNode;
 use Behat\Gherkin\Node\TableNode;
-use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 use Neos\ContentRepository\Core\Factory\ContentRepositoryServiceFactoryDependencies;
 use Neos\ContentRepository\Core\Factory\ContentRepositoryServiceFactoryInterface;
 use Neos\ContentRepository\Core\Factory\ContentRepositoryServiceInterface;
 use Neos\ContentRepository\Core\Feature\NodeModification\Dto\PropertyValuesToWrite;
 use Neos\ContentRepository\Core\NodeType\NodeTypeName;
 use Neos\ContentRepository\Core\Projection\CatchUpOptions;
+use Neos\ContentRepository\Core\Projection\ContentGraph\ContentGraphReadModelInterface;
 use Neos\ContentRepository\Core\Projection\ContentGraph\Filter\FindSubtreeFilter;
 use Neos\ContentRepository\Core\Projection\ContentGraph\Filter\NodeType\NodeTypeCriteria;
 use Neos\ContentRepository\Core\Projection\ContentGraph\Subtree;
 use Neos\ContentRepository\Core\Projection\ContentGraph\VisibilityConstraints;
-use Neos\ContentRepository\Core\SharedModel\Workspace\Workspace;
-use Neos\ContentRepository\Core\Service\ContentStreamPruner;
 use Neos\ContentRepository\Core\Service\ContentStreamPrunerFactory;
 use Neos\ContentRepository\Core\SharedModel\Node\NodeAggregateId;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
-use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamStatus;
 use Neos\ContentRepository\Core\SharedModel\Workspace\WorkspaceName;
 use Neos\ContentRepository\TestSuite\Behavior\Features\Bootstrap\Features\ContentStreamClosing;
 use Neos\ContentRepository\TestSuite\Behavior\Features\Bootstrap\Features\NodeCopying;
@@ -129,14 +127,22 @@ protected function readPayloadTable(TableNode $payloadTable): array
         return $eventPayload;
     }
 
+    /**
+     * @Then /^I expect the content stream "([^"]*)" to exist$/
+     */
+    public function iExpectTheContentStreamToExist(string $rawContentStreamId): void
+    {
+        $contentStream = $this->currentContentRepository->findContentStreamById(ContentStreamId::fromString($rawContentStreamId));
+        Assert::assertNotNull($contentStream, sprintf('Content stream "%s" was expected to exist, but it does not', $rawContentStreamId));
+    }
+
     /**
      * @Then /^I expect the content stream "([^"]*)" to not exist$/
      */
-    public function iExpectTheContentStreamToNotExist(string $rawContentStreamId): void
+    public function iExpectTheContentStreamToNotExist(string $rawContentStreamId, string $not = ''): void
     {
         $contentStream = $this->currentContentRepository->findContentStreamById(ContentStreamId::fromString($rawContentStreamId));
-        $contentStreamExists = $contentStream !== null && !$contentStream->removed;
-        Assert::assertFalse($contentStreamExists, sprintf('Content stream "%s" was not expected to exist, but it does', $rawContentStreamId));
+        Assert::assertNull($contentStream, sprintf('Content stream "%s" was not expected to exist, but it does', $rawContentStreamId));
     }
 
     /**
@@ -245,47 +251,26 @@ protected function getRootNodeAggregateId(): ?NodeAggregateId
     }
 
     /**
-     * @Then the content stream :contentStreamId has status :expectedState
-     */
-    public function theContentStreamHasStatus(string $contentStreamId, string $expectedStatus): void
-    {
-        $contentStream = $this->currentContentRepository->findContentStreamById(ContentStreamId::fromString($contentStreamId));
-        if ($contentStream === null) {
-            Assert::fail(sprintf('Expected content stream "%s" to have status "%s" but it does not exist', $contentStreamId, $expectedStatus));
-        }
-        Assert::assertSame(ContentStreamStatus::tryFrom($expectedStatus), $contentStream->status);
-    }
-
-    /**
-     * @Then the current content stream has status :expectedStatus
+     * @When I prune removed content streams from the event stream
      */
-    public function theCurrentContentStreamHasStatus(string $expectedStatus): void
+    public function iPruneRemovedContentStreamsFromTheEventStream(): void
     {
-        $this->theContentStreamHasStatus(
-            $this->currentContentRepository
-                ->findWorkspaceByName($this->currentWorkspaceName)
-                ->currentContentStreamId->value,
-            $expectedStatus
-        );
+        $this->getContentRepositoryService(new ContentStreamPrunerFactory())->pruneRemovedFromEventStream(fn () => null);
     }
 
     /**
-     * @When I prune unused content streams
+     * @When I expect the content stream pruner status output:
      */
-    public function iPruneUnusedContentStreams(): void
+    public function iExpectTheContentStreamStatus(PyStringNode $pyStringNode): void
     {
-        /** @var ContentStreamPruner $contentStreamPruner */
-        $contentStreamPruner = $this->getContentRepositoryService(new ContentStreamPrunerFactory());
-        $contentStreamPruner->prune();
+        // todo a little dirty to compare the cli output here :D
+        $lines = [];
+        $this->getContentRepositoryService(new ContentStreamPrunerFactory())->outputStatus(function ($line = '') use (&$lines) {
+            $lines[] = $line;
+        });
+        Assert::assertSame($pyStringNode->getRaw(), join("\n", $lines));
     }
 
-    /**
-     * @When I prune removed content streams from the event stream
-     */
-    public function iPruneRemovedContentStreamsFromTheEventStream(): void
-    {
-        $this->getContentRepositoryService(new ContentStreamPrunerFactory())->pruneRemovedFromEventStream();
-    }
 
     abstract protected function getContentRepositoryService(
         ContentRepositoryServiceFactoryInterface $factory
diff --git a/Neos.ContentRepository.TestSuite/Classes/Behavior/Features/Bootstrap/Features/ContentStreamClosing.php b/Neos.ContentRepository.TestSuite/Classes/Behavior/Features/Bootstrap/Features/ContentStreamClosing.php
index 81c21a91567..05c7546c0b2 100644
--- a/Neos.ContentRepository.TestSuite/Classes/Behavior/Features/Bootstrap/Features/ContentStreamClosing.php
+++ b/Neos.ContentRepository.TestSuite/Classes/Behavior/Features/Bootstrap/Features/ContentStreamClosing.php
@@ -15,7 +15,7 @@
 namespace Neos\ContentRepository\TestSuite\Behavior\Features\Bootstrap\Features;
 
 use Behat\Gherkin\Node\TableNode;
-use Neos\ContentRepository\Core\Feature\ContentStreamClosing\Command\CloseContentStream;
+use Neos\ContentRepository\Core\Feature\ContentStreamEventStreamName;
 use Neos\ContentRepository\Core\SharedModel\Workspace\ContentStreamId;
 use Neos\ContentRepository\TestSuite\Behavior\Features\Bootstrap\CRTestSuiteRuntimeVariables;
 
@@ -29,26 +29,17 @@ trait ContentStreamClosing
     abstract protected function readPayloadTable(TableNode $payloadTable): array;
 
     /**
-     * @Given /^the command CloseContentStream is executed with payload:$/
+     * @Given /^the event ContentStreamWasClosed was published with payload:$/
+     * @param TableNode $payloadTable
      * @throws \Exception
      */
-    public function theCommandCloseContentStreamIsExecutedWithPayload(TableNode $payloadTable): void
+    public function theEventContentStreamWasClosedWasPublishedWithPayload(TableNode $payloadTable)
     {
-        $commandArguments = $this->readPayloadTable($payloadTable);
-        $command = CloseContentStream::create(ContentStreamId::fromString($commandArguments['contentStreamId']));
+        $eventPayload = $this->readPayloadTable($payloadTable);
+        $streamName = ContentStreamEventStreamName::fromContentStreamId(
+            ContentStreamId::fromString($eventPayload['contentStreamId'])
+        );
 
-        $this->currentContentRepository->handle($command);
-    }
-
-    /**
-     * @Given /^the command CloseContentStream is executed with payload and exceptions are caught:$/
-     */
-    public function theCommandCloseContentStreamIsExecutedWithPayloadAndExceptionsAreCaught(TableNode $payloadTable): void
-    {
-        try {
-            $this->theCommandCloseContentStreamIsExecutedWithPayload($payloadTable);
-        } catch (\Exception $exception) {
-            $this->lastCommandException = $exception;
-        }
+        $this->publishEvent('ContentStreamWasClosed', $streamName->getEventStreamName(), $eventPayload);
     }
 }
diff --git a/Neos.ContentRepositoryRegistry/Classes/Command/ContentStreamCommandController.php b/Neos.ContentRepositoryRegistry/Classes/Command/ContentStreamCommandController.php
index 40d4330c168..61325fb6a0e 100644
--- a/Neos.ContentRepositoryRegistry/Classes/Command/ContentStreamCommandController.php
+++ b/Neos.ContentRepositoryRegistry/Classes/Command/ContentStreamCommandController.php
@@ -19,55 +19,99 @@ class ContentStreamCommandController extends CommandController
     protected $contentRepositoryRegistry;
 
     /**
-     * Remove all content streams which are not needed anymore from the projections.
+     * Detects if dangling content streams exists and which content streams could be pruned from the event stream
      *
-     * NOTE: This still **keeps** the event stream as is; so it would be possible to re-construct the content stream
-     *       at a later point in time (though we currently do not provide any API for it).
+     * Dangling content streams
+     * ------------------------
      *
-     *       To remove the deleted Content Streams, use `./flow contentStream:pruneRemovedFromEventStream` after running
-     *       `./flow contentStream:prune`.
+     * Content streams that are not removed via the event ContentStreamWasRemoved and are not in use by a workspace
+     * (not a current's workspace content stream).
      *
-     * By default, only content streams in STATE_NO_LONGER_IN_USE and STATE_REBASE_ERROR will be removed.
-     * If you also call with "--removeTemporary", will delete ALL content streams which are currently not assigned
-     * to a workspace (f.e. dangling ones in FORKED or CREATED.).
+     * Previously before Neos 9 beta 15 (#5301), dangling content streams were not removed during publishing, discard or rebase.
+     *
+     * ./flow contentStream:removeDangling
+     *
+     * Pruneable content streams
+     * -------------------------
+     *
+     * Content streams that were removed ContentStreamWasRemoved e.g. after publishing, and are not required for a full
+     * replay to reconstruct the current projections state. The ability to reconstitute a previous state will be lost.
+     *
+     * ./flow contentStream:pruneRemovedFromEventStream
      *
      * @param string $contentRepository Identifier of the content repository. (Default: 'default')
-     * @param boolean $removeTemporary Will delete all content streams which are currently not assigned (Default: false)
      */
-    public function pruneCommand(string $contentRepository = 'default', bool $removeTemporary = false): void
+    public function statusCommand(string $contentRepository = 'default'): void
     {
         $contentRepositoryId = ContentRepositoryId::fromString($contentRepository);
         $contentStreamPruner = $this->contentRepositoryRegistry->buildService($contentRepositoryId, new ContentStreamPrunerFactory());
 
-        $unusedContentStreams = $contentStreamPruner->prune($removeTemporary);
-        $unusedContentStreamsPresent = false;
-        foreach ($unusedContentStreams as $contentStreamId) {
-            $this->outputFormatted('Removed %s', [$contentStreamId->value]);
-            $unusedContentStreamsPresent = true;
-        }
-        if (!$unusedContentStreamsPresent) {
-            $this->outputLine('There are no unused content streams.');
+        $status = $contentStreamPruner->outputStatus(
+            $this->outputLine(...)
+        );
+        if ($status === false) {
+            $this->quit(1);
         }
     }
 
     /**
-     * Remove unused and deleted content streams from the event stream; effectively REMOVING information completely
+     * Removes all nodes, hierarchy relations and content stream entries which are not needed anymore from the projections.
+     *
+     * NOTE: This still **keeps** the event stream as is; so it would be possible to re-construct the content stream at a later point in time.
+     *
+     * HINT: ./flow contentStream:status gives information what is about to be removed
+     *
+     * To prune the removed content streams from the event stream, run ./flow contentStream:pruneRemovedFromEventStream afterwards.
+     *
+     * NOTE: To ensure that no temporary content streams of the *current* moment are removed, a time threshold is configurable via --remove-temporary-before
      *
      * @param string $contentRepository Identifier of the content repository. (Default: 'default')
+     * @param string $removeTemporaryBefore includes all temporary content streams like FORKED or CREATED older than that in the removal. To remove all use --remove-temporary-before=-1sec
      */
-    public function pruneRemovedFromEventStreamCommand(string $contentRepository = 'default'): void
+    public function removeDanglingCommand(string $contentRepository = 'default', string $removeTemporaryBefore = '-1day'): void
     {
         $contentRepositoryId = ContentRepositoryId::fromString($contentRepository);
         $contentStreamPruner = $this->contentRepositoryRegistry->buildService($contentRepositoryId, new ContentStreamPrunerFactory());
 
-        $unusedContentStreams = $contentStreamPruner->pruneRemovedFromEventStream();
-        $unusedContentStreamsPresent = false;
-        foreach ($unusedContentStreams as $contentStreamId) {
-            $this->outputFormatted('Removed events for %s', [$contentStreamId->id->value]);
-            $unusedContentStreamsPresent = true;
+        try {
+            $removeTemporaryBeforeDate = new \DateTimeImmutable($removeTemporaryBefore);
+        } catch (\Exception $exception) {
+            $this->outputLine(sprintf('<error>--remove-temporary-before=%s is not a valid date</error>: %s', $removeTemporaryBefore, $exception->getMessage()));
+            $this->quit(1);
         }
-        if (!$unusedContentStreamsPresent) {
-            $this->outputLine('There are no unused content streams.');
+
+        $now = new \DateTimeImmutable('now');
+        if ($removeTemporaryBeforeDate > $now) {
+            $this->outputLine(sprintf('<error>--remove-temporary-before=%s must be in the past</error>', $removeTemporaryBefore));
+            $this->quit(1);
+        }
+
+        $contentStreamPruner->removeDanglingContentStreams(
+            $this->outputLine(...),
+            $removeTemporaryBeforeDate
+        );
+    }
+
+    /**
+     * Prune removed content streams that are unused from the event stream; effectively REMOVING information completely
+     *
+     * HINT: ./flow contentStream:status gives information what is about to be pruned
+     *
+     * @param string $contentRepository Identifier of the content repository. (Default: 'default')
+     * @param bool $force Prune the unused content streams without confirmation. This cannot be reverted!
+     */
+    public function pruneRemovedFromEventStreamCommand(string $contentRepository = 'default', bool $force = false): void
+    {
+        if (!$force && !$this->output->askConfirmation(sprintf('> This will prune removed content streams that are unused from the event stream in content repository "%s" (see flow contentStream:status). Are you sure to proceed? (y/n) ', $contentRepository), false)) {
+            $this->outputLine('<comment>Abort.</comment>');
+            return;
         }
+
+        $contentRepositoryId = ContentRepositoryId::fromString($contentRepository);
+        $contentStreamPruner = $this->contentRepositoryRegistry->buildService($contentRepositoryId, new ContentStreamPrunerFactory());
+
+        $contentStreamPruner->pruneRemovedFromEventStream(
+            $this->outputLine(...)
+        );
     }
 }
diff --git a/Neos.ContentRepositoryRegistry/Classes/Command/MigrateEventsCommandController.php b/Neos.ContentRepositoryRegistry/Classes/Command/MigrateEventsCommandController.php
index c0401b67d1c..b43c6023464 100644
--- a/Neos.ContentRepositoryRegistry/Classes/Command/MigrateEventsCommandController.php
+++ b/Neos.ContentRepositoryRegistry/Classes/Command/MigrateEventsCommandController.php
@@ -18,6 +18,18 @@ public function __construct(
         parent::__construct();
     }
 
+    /**
+     * Temporary low level backup to ensure the prune migration https://github.com/neos/neos-development-collection/pull/5297 is safe
+     *
+     * @param string $contentRepository Identifier of the Content Repository to backup
+     */
+    public function backupCommand(string $contentRepository = 'default'): void
+    {
+        $contentRepositoryId = ContentRepositoryId::fromString($contentRepository);
+        $eventMigrationService = $this->contentRepositoryRegistry->buildService($contentRepositoryId, $this->eventMigrationServiceFactory);
+        $eventMigrationService->backup($this->outputLine(...));
+    }
+
     /**
      * Migrates initial metadata & roles from the CR core workspaces to the corresponding Neos database tables
      *
diff --git a/Neos.ContentRepositoryRegistry/Classes/Service/EventMigrationService.php b/Neos.ContentRepositoryRegistry/Classes/Service/EventMigrationService.php
index 9cd7d315f80..e88a6139e2b 100644
--- a/Neos.ContentRepositoryRegistry/Classes/Service/EventMigrationService.php
+++ b/Neos.ContentRepositoryRegistry/Classes/Service/EventMigrationService.php
@@ -56,6 +56,14 @@ public function __construct(
     ) {
     }
 
+    public function backup(\Closure $outputFn): void
+    {
+        $backupEventTableName = DoctrineEventStoreFactory::databaseTableName($this->contentRepositoryId)
+            . '_bkp_' . date('Y_m_d_H_i_s');
+        $this->copyEventTable($backupEventTableName);
+        $outputFn(sprintf('Backup. Copied events table to %s', $backupEventTableName));
+    }
+
     /**
      * The following things have to be migrated:
      *
diff --git a/Neos.Neos/Classes/Command/CrCommandController.php b/Neos.Neos/Classes/Command/CrCommandController.php
index ed2ab33e522..d94cbeac1c9 100644
--- a/Neos.Neos/Classes/Command/CrCommandController.php
+++ b/Neos.Neos/Classes/Command/CrCommandController.php
@@ -165,8 +165,7 @@ public function pruneCommand(string $contentRepository = 'default', bool $force
         $this->workspaceService->pruneWorkspaceMetadata($contentRepositoryId);
 
         // reset the events table
-        $contentStreamPruner->pruneAll();
-        $workspaceMaintenanceService->pruneAll();
+        $contentStreamPruner->pruneAllWorkspacesAndContentStreamsFromEventStream();
 
         // reset the projections state
         $projectionService->resetAllProjections();